@Test
public void failureHandlerExecutesFailoverForEachBatchItemSeparately() {

    // given
    Builder builder = createTestObjectFactoryBuilder();
    ClientObjectFactory<JestClient, Bulk> config = builder.build();
    FailoverPolicy failoverPolicy = Mockito.spy(new NoopFailoverPolicy());
    String payload1 = "test1";
    String payload2 = "test2";
    Bulk bulk = new Bulk.Builder()
            .addAction(spy(new Index.Builder(payload1)).build())
            .addAction(spy(new Index.Builder(payload2)).build())
            .build();
    // when
    config.createFailureHandler(failoverPolicy).apply(bulk);
    // then
    ArgumentCaptor<FailedItemSource> captor = ArgumentCaptor.forClass(FailedItemSource.class);
    verify(failoverPolicy, times(2)).deliver(captor.capture());
    assertTrue(captor.getAllValues().get(0).getSource().equals(payload1));
    assertTrue(captor.getAllValues().get(1).getSource().equals(payload2));
}