private void closeWriters(String source, Map<String, AzureBlobWriter> sourceWriterMap) throws Exception {
    Set<CompletableFuture<Void>> pendingClose = ConcurrentHashMap.newKeySet();
    try {
      sourceWriterMap.forEach((stream, writer) -> {
        LOG.info("Closing topic:{}", stream);
        CompletableFuture<Void> future = CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
              try {
                writer.close();
              } catch (Exception e) {
                throw new SystemProducerException("Close failed for topic " + stream, e);
              }
            }
          }, asyncBlobThreadPool);
        pendingClose.add(future);
        future.handle((aVoid, throwable) -> {
          sourceWriterMap.remove(writer);
          if (throwable != null) {
            throw new SystemProducerException("Close failed for topic " + stream, throwable);
          } else {
            LOG.info("Blob close finished for stream " + stream);
            return aVoid;
          }
        });
      });
      CompletableFuture<Void> future = CompletableFuture.allOf(pendingClose.toArray(new CompletableFuture[0]));
      LOG.info("Flush source: {} has pending closes: {} ", source, pendingClose.size());
      future.get((long) closeTimeout, TimeUnit.MILLISECONDS);
    } finally {
      pendingClose.clear();
    }
  }
// 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