@Test
public void failureHandlerExecutesFailoverForEachBatchItemSeparately() {

    // given
    Builder builder = createTestObjectFactoryBuilder();
    ClientObjectFactory<TransportClient, BulkRequest> config = builder.build();
    FailoverPolicy failoverPolicy = spy(new NoopFailoverPolicy());
    String payload1 = "test1";
    String payload2 = "test2";
    BulkRequest bulk = new BulkRequest()
            .add(spy(new IndexRequest().source(payload1, XContentType.CBOR)))
            .add(spy(new IndexRequest().source(payload2, XContentType.CBOR)));
    // when
    config.createFailureHandler(failoverPolicy).apply(bulk);
    // then
    ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
    verify(failoverPolicy, times(2)).deliver(captor.capture());
    assertTrue(captor.getAllValues().contains(payload1));
    assertTrue(captor.getAllValues().contains(payload2));
}