@VisibleForTesting
  AzureBlobWriter getOrCreateWriter(String source, OutgoingMessageEnvelope messageEnvelope) {
    String writerMapKey;
    String blobURLPrefix;
    String partitionKey = getPartitionKey(messageEnvelope);
    // using most significant bits in UUID (8 digits) to avoid collision in blob names
    if (partitionKey == null) {
      writerMapKey = messageEnvelope.getSystemStream().getStream();
      blobURLPrefix = String.format(BLOB_NAME_PREFIX, messageEnvelope.getSystemStream().getStream());
    } else {
      writerMapKey = messageEnvelope.getSystemStream().getStream() + "/" + partitionKey;
      blobURLPrefix = String.format(BLOB_NAME_PARTITION_PREFIX, messageEnvelope.getSystemStream().getStream(), partitionKey);
    }
    Map<String, AzureBlobWriter> sourceWriterMap = writerMap.get(source);
    if (sourceWriterMap == null) {
      throw new SystemProducerException("Attempting to send to source: " + source + " but it is not registered");
    }
    AzureBlobWriter writer = sourceWriterMap.get(writerMapKey);
    if (writer == null) {
      synchronized (sourceWriterCreationLockMap.get(source)) {
        writer = sourceWriterMap.get(writerMapKey);
        if (writer == null) {
          AzureBlobWriterMetrics writerMetrics =
              new AzureBlobWriterMetrics(metrics.getAggregateMetrics(), metrics.getSystemMetrics(), metrics.getSourceMetrics(source));
          writer = createNewWriter(blobURLPrefix, writerMetrics, messageEnvelope.getSystemStream().getStream());
          sourceWriterMap.put(writerMapKey, writer);
        }
      }
    }
    return writer;
  }
// 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