@Test
public void failureHandlerExecutesFailoverForEachBatchItemSeparately() {

    // given
    BufferedJestHttpObjectFactory.Builder builder = createTestObjectFactoryBuilder();
    ClientObjectFactory<JestClient, Bulk> config = builder.build();
    FailoverPolicy failoverPolicy = Mockito.spy(new NoopFailoverPolicy());
    String payload1 = "test1";
    String payload2 = "test2";
    ItemSource<ByteBuf> source1 = createDefaultTestBuffereItemSource(payload1);
    ItemSource<ByteBuf> source2 = createDefaultTestBuffereItemSource(payload2);
    Bulk bulk = createTestBatch(source1, source2);
    // when
    config.createFailureHandler(failoverPolicy).apply(bulk);
    // then
    ArgumentCaptor<FailedItemSource> captor = ArgumentCaptor.forClass(FailedItemSource.class);
    verify(failoverPolicy, times(2)).deliver(captor.capture());
    FailedItemSource<ByteBuf> item1 = captor.getAllValues().get(0);
    assertTrue(item1.getSource().toString(Charset.defaultCharset()).equals(payload1));
    FailedItemSource<ByteBuf> item2 = captor.getAllValues().get(1);
    assertTrue(item2.getSource().toString(Charset.defaultCharset()).equals(payload2));
}