### Question:
HCHttp implements ClientObjectFactory<HttpClient, BatchRequest> { @Override public Function<BatchRequest, Boolean> createFailureHandler(FailoverPolicy failover) { return batchRequest -> { long start = System.currentTimeMillis(); int batchSize = batchRequest.getIndexRequests().size(); LOG.warn("BatchRequest of {} indexRequests failed. Redirecting to {}", batchSize, failover.getClass().getName()); batchRequest.getIndexRequests().forEach(indexRequest -> { try { failover.deliver(failedItemOps.createItem(indexRequest)); } catch (Exception e) { LOG.error(e.getMessage(), e); } }); LOG.trace("BatchRequest of {} indexRequests redirected in {} ms", batchSize, System.currentTimeMillis() - start); return true; }; } HCHttp(Builder builder); @Override Function<BatchRequest, Boolean> createFailureHandler(FailoverPolicy failover); @Override BatchOperations<BatchRequest> createBatchOperations(); @Override Collection<String> getServerList(); @Override HttpClient createClient(); @Override Function<BatchRequest, Boolean> createBatchListener(FailoverPolicy failoverPolicy); @Override void execute(IndexTemplate indexTemplate); @Override void addOperation(Operation operation); @PluginBuilderFactory static HCHttp.Builder newBuilder(); @Override void start(); @Override void stop(); @Override boolean isStarted(); @Override boolean isStopped(); static final String PLUGIN_NAME; }### Answer:
@Test public void failureHandlerExecutesFailoverForEachBatchItemSeparately() { HCHttp.Builder builder = createDefaultHttpObjectFactoryBuilder(); ClientObjectFactory<HttpClient, BatchRequest> 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); BatchRequest.Builder batchBuilder = spy(new BatchRequest.Builder()); BatchRequest request = createTestBatch(batchBuilder, source1, source2); config.createFailureHandler(failoverPolicy).apply(request); ArgumentCaptor<FailedItemSource> captor = ArgumentCaptor.forClass(FailedItemSource.class); verify(failoverPolicy, times(2)).deliver(captor.capture()); assertTrue(captor.getAllValues().get(0).getSource().equals(source1.getSource())); assertTrue(captor.getAllValues().get(1).getSource().equals(source2.getSource())); }