### Question:
BulkProcessorObjectFactory implements ClientObjectFactory<TransportClient, BulkRequest> { @Override public Function<BulkRequest, Boolean> createFailureHandler(FailoverPolicy failover) { return new Function<BulkRequest, Boolean>() { private final BulkRequestIntrospector introspector = new BulkRequestIntrospector(); @Override public Boolean apply(BulkRequest bulk) { introspector.items(bulk).forEach(failedItem -> failover.deliver(failedItem)); return true; } }; } protected  BulkProcessorObjectFactory(Collection<String> serverUris, Auth auth); protected  BulkProcessorObjectFactory(Collection<String> serverUris, Auth auth, ClientSettings clientSettings); @Override Collection<String> getServerList(); @Override TransportClient createClient(); @Override Function<BulkRequest, Boolean> createBatchListener(FailoverPolicy failoverPolicy); @Override Function<BulkRequest, Boolean> createFailureHandler(FailoverPolicy failover); @Override BatchOperations<BulkRequest> createBatchOperations(); @Override void execute(IndexTemplate indexTemplate); @Override void addOperation(Operation operation); @PluginBuilderFactory static Builder newBuilder();  }### Answer:
@Test public void failureHandlerExecutesFailoverForEachBatchItemSeparately() { 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))); config.createFailureHandler(failoverPolicy).apply(bulk); ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class); verify(failoverPolicy, times(2)).deliver(captor.capture()); assertTrue(captor.getAllValues().contains(payload1)); assertTrue(captor.getAllValues().contains(payload2)); }
@Test public void failureHandlerExecutesFailoverForEachBatchItemSeparately() { 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))) .add(spy(new IndexRequest().source(payload2))); config.createFailureHandler(failoverPolicy).apply(bulk); ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class); verify(failoverPolicy, times(2)).deliver((String) captor.capture()); assertTrue(captor.getAllValues().contains(payload1)); assertTrue(captor.getAllValues().contains(payload2)); }