### Question:
JestHttpObjectFactory implements ClientObjectFactory<JestClient, Bulk> { @Override public Function<Bulk, Boolean> createBatchListener(FailoverPolicy failoverPolicy) { return new Function<Bulk, Boolean>() { private Function<Bulk, Boolean> failureHandler = createFailureHandler(failoverPolicy); @Override public Boolean apply(Bulk bulk) { while (!operations.isEmpty()) { try { operations.remove().execute(); } catch (Exception e) { LOG.error("Deferred operation failed: {}", e.getMessage()); } } if (backoffPolicy.shouldApply(bulk)) { LOG.warn("Backoff applied. Request rejected."); failureHandler.apply(bulk); return false; } else { backoffPolicy.register(bulk); } JestResultHandler<JestResult> jestResultHandler = createResultHandler(bulk, failureHandler); createClient().executeAsync(bulk, jestResultHandler); 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 failoverHandlerIsExecutedImmediatelyIfBackoffPolicyShouldApply() { BackoffPolicy<AbstractAction<BulkResult>> backoffPolicy = mock(BackoffPolicy.class); when(backoffPolicy.shouldApply(any())).thenReturn(true); Builder builder = createTestObjectFactoryBuilder(); builder.withBackoffPolicy(backoffPolicy); JestHttpObjectFactory config = spy(builder.build()); String payload1 = "test1"; Bulk bulk = createTestBatch(payload1); FailoverPolicy failoverPolicy = mock(FailoverPolicy.class); Function<Bulk, Boolean> listener = config.createBatchListener(failoverPolicy); listener.apply(bulk); ArgumentCaptor<FailedItemSource> captor = ArgumentCaptor.forClass(FailedItemSource.class); verify(failoverPolicy, times(1)).deliver(captor.capture()); assertEquals(payload1, captor.getValue().getSource()); }
@Test public void failoverHandlerIsNotExecutedImmediatelyIfBackoffPolicyShouldNotApply() { BackoffPolicy<AbstractAction<BulkResult>> backoffPolicy = mock(BackoffPolicy.class); when(backoffPolicy.shouldApply(any())).thenReturn(false); Builder builder = createTestObjectFactoryBuilder(); builder.withBackoffPolicy(backoffPolicy); JestHttpObjectFactory config = spy(builder.build()); String payload1 = "test1"; Bulk bulk = createTestBatch(payload1); FailoverPolicy failoverPolicy = mock(FailoverPolicy.class); Function<Bulk, Boolean> listener = config.createBatchListener(failoverPolicy); listener.apply(bulk); ArgumentCaptor<FailedItemSource> captor = ArgumentCaptor.forClass(FailedItemSource.class); verify(failoverPolicy, never()).deliver(captor.capture()); }