AzureBlobOutputStream extends OutputStream { @Override public synchronized void flush() { if (byteArrayOutputStream.isPresent()) { uploadBlockAsync(); } } 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 testFlush() throws Exception { azureBlobOutputStream.write(BYTES); azureBlobOutputStream.flush(); azureBlobOutputStream.close(); int blockNum = 0; String blockId = String.format("%05d", blockNum); String blockIdEncoded = Base64.getEncoder().encodeToString(blockId.getBytes()); verify(mockCompression).compress(BYTES); ArgumentCaptor<ByteBuffer> argument = ArgumentCaptor.forClass(ByteBuffer.class); verify(azureBlobOutputStream).stageBlock(eq(blockIdEncoded(0)), argument.capture(), eq((int) COMPRESSED_BYTES.length)); Assert.assertEquals(ByteBuffer.wrap(COMPRESSED_BYTES), argument.getAllValues().get(0)); verify(mockMetrics).updateAzureUploadMetrics(); }