AzureBlobAvroWriter implements AzureBlobWriter { @VisibleForTesting byte[] encodeRecord(IndexedRecord record) { ByteArrayOutputStream out = new ByteArrayOutputStream(); Schema schema = record.getSchema(); try { EncoderFactory encoderfactory = new EncoderFactory(); BinaryEncoder encoder = encoderfactory.binaryEncoder(out, null); DatumWriter<IndexedRecord> writer; if (record instanceof SpecificRecord) { writer = new SpecificDatumWriter<>(schema); } else { writer = new GenericDatumWriter<>(schema); } writer.write(record, encoder); encoder.flush(); } catch (Exception e) { throw new SamzaException("Unable to serialize Avro record using schema within the record: " + schema.toString(), e); } return out.toByteArray(); } 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 testEncodeRecord() throws Exception { azureBlobAvroWriter = spy(new AzureBlobAvroWriter(PowerMockito.mock(BlobContainerAsyncClient.class), mock(AzureBlobWriterMetrics.class), threadPool, THRESHOLD, 60000, "test", mockDataFileWriter, mockAzureBlobOutputStream, mockBlockBlobAsyncClient, blobMetadataGeneratorFactory, blobMetadataGeneratorConfig, STREAM_NAME, Long.MAX_VALUE, Long.MAX_VALUE, mockCompression, false)); IndexedRecord record = new GenericRecordEvent(); Assert.assertTrue(Arrays.equals(encodeRecord(record), azureBlobAvroWriter.encodeRecord(record))); }
@Test public void testMultipleThreadWrites() throws Exception { Thread t1 = writeInThread(ome, azureBlobAvroWriter, 10); OutgoingMessageEnvelope ome2 = createOMEGenericRecord("TOPIC2"); Thread t2 = writeInThread(ome2, azureBlobAvroWriter, 10); t1.start(); t2.start(); t1.join(60000); t2.join(60000); verify(mockDataFileWriter, times(10)).appendEncoded(ByteBuffer.wrap(encodedRecord)); verify(mockDataFileWriter, times(10)).appendEncoded(ByteBuffer.wrap(encodeRecord((IndexedRecord) ome2.getMessage()))); verify(mockAzureBlobOutputStream, times(20)).incrementNumberOfRecordsInBlob(); }