AzureBlobOutputStream extends OutputStream { @Override public synchronized void write(int b) { if (!byteArrayOutputStream.isPresent()) { throw new IllegalStateException("Internal Buffer must have been released earlier for blob " + blobAsyncClient.getBlobUrl().toString()); } if (byteArrayOutputStream.get().size() + 1 > maxBlockFlushThresholdSize) { uploadBlockAsync(); } byteArrayOutputStream.get().write(b); metrics.updateWriteByteMetrics(1); } AzureBlobOutputStream(BlockBlobAsyncClient blobAsyncClient, Executor blobThreadPool, AzureBlobWriterMetrics metrics,
      BlobMetadataGeneratorFactory blobMetadataGeneratorFactory, Config blobMetadataGeneratorConfig, String streamName,
      long flushTimeoutMs, int maxBlockFlushThresholdSize, Compression compression); @VisibleForTesting  AzureBlobOutputStream(BlockBlobAsyncClient blobAsyncClient, Executor blobThreadPool, AzureBlobWriterMetrics metrics,
      BlobMetadataGeneratorFactory blobMetadataGeneratorFactory, Config blobMetadataGeneratorConfig, String streamName,
      long flushTimeoutMs, int maxBlockFlushThresholdSize,
      ByteArrayOutputStream byteArrayOutputStream, Compression compression); @Override synchronized void write(int b); @Override synchronized void write(byte[] b, int off, int len); @Override synchronized void flush(); @Override synchronized void close(); synchronized long getSize(); synchronized void releaseBuffer(); synchronized void incrementNumberOfRecordsInBlob();  }
@Test public void testWrite() { byte[] b = new byte[THRESHOLD - 10]; azureBlobOutputStream.write(b, 0, THRESHOLD - 10); verify(azureBlobOutputStream, never()).stageBlock(anyString(), any(ByteBuffer.class), anyInt()); verify(mockMetrics).updateWriteByteMetrics(THRESHOLD - 10); verify(mockMetrics, never()).updateAzureUploadMetrics(); }