AzureBlobSystemProducer implements SystemProducer { @Override public synchronized void stop() { if (!isStarted) { LOG.warn("Attempting to stop a producer that was not started."); return; } if (isStopped) { LOG.warn("Attempting to stop an already stopped producer."); return; } try { writerMap.forEach((source, sourceWriterMap) -> flush(source)); asyncBlobThreadPool.shutdown(); isStarted = false; } catch (Exception e) { throw new SystemProducerException("Stop failed with exception.", e); } finally { writerMap.clear(); isStopped = true; } } AzureBlobSystemProducer(String systemName, AzureBlobConfig config, MetricsRegistry metricsRegistry); @Override synchronized void start(); @Override synchronized void stop(); @Override synchronized void register(String source); @Override void send(String source, OutgoingMessageEnvelope messageEnvelope); @Override void flush(String source);  }
@Test public void testStop() throws Exception { doNothing().when(mockAzureWriter).close(); systemProducer.register(SOURCE); systemProducer.start(); systemProducer.send(SOURCE, ome); systemProducer.flush(SOURCE); systemProducer.stop(); verify(mockAzureWriter).flush(); verify(mockAzureWriter).close(); }