AzureBlobSystemProducer implements SystemProducer { @Override public void flush(String source) { if (!isStarted) { throw new SystemProducerException("Trying to flush before producer has started."); } if (isStopped) { throw new SystemProducerException("Flushing after producer has been stopped."); } ReadWriteLock lock = sourceSendFlushLockMap.get(source); if (lock == null) { throw new SystemProducerException("Attempting to flush source: " + source + " but it was not registered"); } lock.writeLock().lock(); Map<String, AzureBlobWriter> sourceWriterMap = writerMap.get(source); try { flushWriters(sourceWriterMap); closeWriters(source, sourceWriterMap); } catch (Exception e) { metrics.updateErrorMetrics(source); throw new SystemProducerException("Flush failed for system:" + systemName + " and source: " + source, e); } finally { sourceWriterMap.clear(); lock.writeLock().unlock(); } } 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 testFlush() throws IOException { systemProducer.register(SOURCE); systemProducer.start(); systemProducer.send(SOURCE, ome); systemProducer.flush(SOURCE); verify(mockAzureWriter).flush(); verify(mockAzureWriter).close(); }
@Test(expected = SystemProducerException.class) public void testFlushBeforeStart() throws Exception { systemProducer.flush(SOURCE); }