@VisibleForTesting
  AzureBlobWriter createNewWriter(String blobURL, AzureBlobWriterMetrics writerMetrics, String streamName) {
    try {
      return writerFactory.getWriterInstance(containerAsyncClient, blobURL, asyncBlobThreadPool, writerMetrics,
          blobMetadataGeneratorFactory, blobMetadataGeneratorConfig, streamName,
          blockFlushThresholdSize, flushTimeoutMs,
          CompressionFactory.getInstance().getCompression(config.getCompressionType(systemName)),
          config.getSuffixRandomStringToBlobName(systemName),
          config.getMaxBlobSize(systemName),
          config.getMaxMessagesPerBlob(systemName),
          config.getInitBufferSizeBytes(systemName));
    } catch (Exception e) {
      throw new RuntimeException("Failed to create a writer for the producer.", 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