ElasticsearchSystemProducer implements SystemProducer { @Override public void send(String source, OutgoingMessageEnvelope envelope) { IndexRequest indexRequest = indexRequestFactory.getIndexRequest(envelope); sourceBulkProcessor.get(source).add(indexRequest); } 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 testSend() throws Exception { OutgoingMessageEnvelope envelope = mock(OutgoingMessageEnvelope.class); IndexRequest indexRequest = mock(IndexRequest.class); when(INDEX_REQUEST_FACTORY.getIndexRequest(envelope)).thenReturn(indexRequest); producer.send(SOURCE_ONE, envelope); verify(processorOne).add(indexRequest); }