private void flushWriters(Map<String, AzureBlobWriter> sourceWriterMap) {
    sourceWriterMap.forEach((stream, writer) -> {
      try {
        LOG.info("Flushing topic:{}", stream);
        writer.flush();
      } catch (IOException e) {
        throw new SystemProducerException("Close failed for topic " + stream, e);
      }
    });
  }
// find the writer in the writerMap else create one
@param source for which to find/create the writer
@param messageEnvelope to fetch the schema from if writer needs to be created
@return an AzureBlobWriter object
samza-azure/src/main/java/org/apache/samza/system/azureblob/producer/AzureBlobSystemProducer.java