@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);
        }
      }
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