private void startNextBlob(Optional<GenericContainer> optionalGenericContainer) throws IOException {
    if (currentBlobWriterComponents != null) {
      LOG.info("Starting new blob as current blob size is "
          + currentBlobWriterComponents.azureBlobOutputStream.getSize()
          + " and max blob size is " + maxBlobSize
          + " or number of records is " + recordsInCurrentBlob
          + " and max records in blob is " + maxRecordsPerBlob);
      currentBlobWriterComponents.dataFileWriter.flush();
      currentBlobWriterComponents.azureBlobOutputStream.releaseBuffer();
      recordsInCurrentBlob = 0;
    }
    // datumWriter is null when AzureBlobAvroWriter is created but has not yet received a message.
    // optionalIndexedRecord is the first message in this case.
    if (datumWriter == null) {
      if (optionalGenericContainer.isPresent()) {
        GenericContainer container = optionalGenericContainer.get();
        schema = container.getSchema();
        if (container instanceof SpecificRecord) {
          datumWriter = new SpecificDatumWriter<>(schema);
        } else {
          datumWriter = new GenericDatumWriter<>(schema);
        }
      } else {
        throw new IllegalStateException("Writing without schema setup.");
      }
    }
    String blobURL;
    if (useRandomStringInBlobName) {
      blobURL = String.format(BLOB_NAME_RANDOM_STRING_AVRO, blobURLPrefix,
          UTC_FORMATTER.format(System.currentTimeMillis()), UUID.randomUUID().toString().substring(0, 8),
          compression.getFileExtension());
    } else {
      blobURL = String.format(BLOB_NAME_AVRO, blobURLPrefix,
          UTC_FORMATTER.format(System.currentTimeMillis()), compression.getFileExtension());
    }
    LOG.info("Creating new blob: {}", blobURL);
    BlockBlobAsyncClient blockBlobAsyncClient = containerAsyncClient.getBlobAsyncClient(blobURL).getBlockBlobAsyncClient();

    DataFileWriter<Object> dataFileWriter = new DataFileWriter<>(datumWriter);
    AzureBlobOutputStream azureBlobOutputStream;
    try {
      azureBlobOutputStream = new AzureBlobOutputStream(blockBlobAsyncClient, blobThreadPool, metrics,
          blobMetadataGeneratorFactory, blobMetadataGeneratorConfig,
          streamName, flushTimeoutMs, maxBlockFlushThresholdSize, compression, initBufferSize);
    } catch (Exception e) {
      throw new SamzaException("Unable to create AzureBlobOutputStream", e);
    }
    dataFileWriter.create(schema, azureBlobOutputStream);
    dataFileWriter.setFlushOnEveryBlock(false);
    this.currentBlobWriterComponents = new BlobWriterComponents(dataFileWriter, azureBlobOutputStream, blockBlobAsyncClient);
    allBlobWriterComponents.add(this.currentBlobWriterComponents);
    LOG.info("Created new blob: {}", blobURL);
  }
This method closes all DataFileWriters and output streams associated with all the blobs created.
flush should be explicitly called before close.
{@inheritDoc}
@throws IllegalStateException when closing a closed writer
@throws SamzaException if underlying DataFileWriter.close fails
samza-azure/src/main/java/org/apache/samza/system/azureblob/avro/AzureBlobAvroWriter.java