### Question:
JestHttpObjectFactory implements ClientObjectFactory<JestClient, Bulk> { @Override public Function<Bulk, Boolean> createFailureHandler(FailoverPolicy failover) { return new Function<Bulk, Boolean>() { private final JestBatchIntrospector introspector = new JestBatchIntrospector(); @Override public Boolean apply(Bulk bulk) { Collection items = introspector.items(bulk); LOG.warn(String.format("Batch of %s items failed. Redirecting to %s", items.size(), failover.getClass().getName())); items.forEach(item -> { Index failedAction = (Index) item; failover.deliver(failedItemOps.createItem(failedAction)); }); return true; } }; } protected  JestHttpObjectFactory(Builder builder); @Override Collection<String> getServerList(); @Override JestClient createClient(); @Override Function<Bulk, Boolean> createBatchListener(FailoverPolicy failoverPolicy); @Override Function<Bulk, Boolean> createFailureHandler(FailoverPolicy failover); @Override BatchOperations<Bulk> createBatchOperations(); @Override void execute(IndexTemplate indexTemplate); @Override void addOperation(Operation operation); @PluginBuilderFactory static Builder newBuilder(); @Override void start(); @Override void stop(); @Override boolean isStarted(); @Override boolean isStopped();  }### Answer:
@Test public void failureHandlerExecutesFailoverForEachBatchItemSeparately() { 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(); config.createFailureHandler(failoverPolicy).apply(bulk); 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)); }