ElasticsearchSystemProducer implements SystemProducer { @Override public void flush(String source) { sourceBulkProcessor.get(source).flush(); if (sendFailed.get()) { String message = String.format("Unable to send message from %s to system %s.", source, system); LOGGER.error(message); Throwable cause = thrown.get(); if (cause != null) { throw new SamzaException(message, cause); } else { throw new SamzaException(message); } } LOGGER.info(String.format("Flushed %s to %s.", source, system)); } ElasticsearchSystemProducer(String system, BulkProcessorFactory bulkProcessorFactory,
                                     Client client, IndexRequestFactory indexRequestFactory,
                                     ElasticsearchSystemProducerMetrics metrics); @Override void start(); @Override void stop(); @Override void register(final String source); @Override void send(String source, OutgoingMessageEnvelope envelope); @Override void flush(String source);  }
@Test public void testFlushNoFailedSend() throws Exception { producer.flush(SOURCE_ONE); verify(processorOne).flush(); verify(processorTwo, never()).flush(); }