private synchronized void uploadBlockAsync() {
    if (!byteArrayOutputStream.isPresent()) {
      return;
    }
    long size = byteArrayOutputStream.get().size();
    if (size == 0) {
      return;
    }
    LOG.info("Blob: {} uploadBlock. Size:{}", blobAsyncClient.getBlobUrl().toString(), size);

    // Azure sdk requires block Id to be encoded and all blockIds of a blob to be of the same length
    // also, a block blob can have upto 50,000 blocks, hence using a 5 digit block id.
    String blockId = String.format("%05d", blockNum);
    String blockIdEncoded = Base64.getEncoder().encodeToString(blockId.getBytes());
    blockList.add(blockIdEncoded);
    byte[] localByte = byteArrayOutputStream.get().toByteArray();
    byteArrayOutputStream.get().reset();
    totalUploadedBlockSize += localByte.length;

    CompletableFuture<Void> future = CompletableFuture.runAsync(new Runnable() {
      // call async stageblock and add to future
      @Override
      public void run() {
        byte[] compressedLocalByte = compression.compress(localByte);
        int blockSize = compressedLocalByte.length;

        try {
          ByteBuffer outputStream = ByteBuffer.wrap(compressedLocalByte, 0, blockSize);
          metrics.updateCompressByteMetrics(blockSize);
          LOG.info("{} Upload block start for blob: {} for block size:{}.", blobAsyncClient.getBlobUrl().toString(), blockId, blockSize);
          metrics.updateAzureUploadMetrics();
          // StageBlock generates exception on Failure.
          stageBlock(blockIdEncoded, outputStream, blockSize);
        } catch (Exception e) {
          String msg = String.format("Upload block for blob: %s failed for blockid: %s.", blobAsyncClient.getBlobUrl().toString(), blockId);
          LOG.error(msg, e);
          throw new AzureException(msg, e);
        }
      }
    }, blobThreadPool);

    pendingUpload.add(future);

    future.handle((aVoid, throwable) -> {
      if (throwable == null) {
        LOG.info("Upload block for blob: {} with blockid: {} finished.", blobAsyncClient.getBlobUrl().toString(), blockId);
        pendingUpload.remove(future);
        return aVoid;
      } else {
        throw new AzureException("Blob upload failed for blob " + blobAsyncClient.getBlobUrl().toString()
            + " and block with id: " + blockId, throwable);
      }
    });

    blockNum += 1;
    if (blockNum >= MAX_BLOCKS_IN_AZURE_BLOB) {
      throw new AzureException("Azure blob only supports 50000 blocks in a blob. Current number of blocks is " + blockNum);
    }
  }
This api will async upload the outputstream into block using stageBlocks,
reint outputstream
and add the operation to future.
@throws RuntimeException when
           - blob's stageBlock fails after MAX_ATTEMPTs
           - number of blocks exceeds MAX_BLOCKS_IN_AZURE_BLOB
samza-azure/src/main/java/org/apache/samza/system/azureblob/avro/AzureBlobOutputStream.java