AzureBlobAvroWriter implements AzureBlobWriter { @Override public void close() { synchronized (currentDataFileWriterLock) { if (isClosed) { throw new IllegalStateException("Attempting to close an already closed AzureBlobAvroWriter"); } allBlobWriterComponents.forEach(blobWriterComponents -> { try { closeDataFileWriter(blobWriterComponents.dataFileWriter, blobWriterComponents.azureBlobOutputStream, blobWriterComponents.blockBlobAsyncClient); } catch (IOException e) { throw new SamzaException(e); } }); isClosed = true; } } AzureBlobAvroWriter(BlobContainerAsyncClient containerAsyncClient, String blobURLPrefix,
      Executor blobThreadPool, AzureBlobWriterMetrics metrics,
      BlobMetadataGeneratorFactory blobMetadataGeneratorFactory, Config blobMetadataGeneratorConfig, String streamName,
      int maxBlockFlushThresholdSize, long flushTimeoutMs, Compression compression, boolean useRandomStringInBlobName,
      long maxBlobSize, long maxRecordsPerBlob); @VisibleForTesting  AzureBlobAvroWriter(BlobContainerAsyncClient containerAsyncClient, AzureBlobWriterMetrics metrics,
      Executor blobThreadPool, int maxBlockFlushThresholdSize, int flushTimeoutMs, String blobURLPrefix,
      DataFileWriter<IndexedRecord> dataFileWriter,
      AzureBlobOutputStream azureBlobOutputStream, BlockBlobAsyncClient blockBlobAsyncClient,
      BlobMetadataGeneratorFactory blobMetadataGeneratorFactory, Config blobMetadataGeneratorConfig, String streamName,
      long maxBlobSize, long maxRecordsPerBlob, Compression compression, boolean useRandomStringInBlobName); @Override void write(OutgoingMessageEnvelope ome); @Override void flush(); @Override void close();  }
@Test public void testClose() throws Exception { azureBlobAvroWriter.close(); verify(mockDataFileWriter).close(); }
@Test(expected = RuntimeException.class) public void testCloseWhenOutputStreamFails() throws Exception { doThrow(new IOException("DataFileWriter failed")).when(mockDataFileWriter).close(); doThrow(new RuntimeException("failed")).when(mockAzureBlobOutputStream).close(); azureBlobAvroWriter.close(); }