public void write(
            ByteBuffer record,
            TieredStorageSubpartitionId subpartitionId,
            Buffer.DataType dataType,
            boolean isBroadcast)
            throws IOException {

        if (isBroadcast && !isBroadcastOnly) {
            int currentPosition = record.position();
            for (int i = 0; i < numSubpartitions; ++i) {
                // As the tiered storage subpartition ID is created only for broadcast records,
                // which are fewer than normal records, the performance impact of generating new
                // TieredStorageSubpartitionId objects is expected to be manageable. If the
                // performance is significantly affected, this logic will be optimized accordingly.
                bufferAccumulator.receive(
                        record, new TieredStorageSubpartitionId(i), dataType, isBroadcast);
                record.position(currentPosition);
            }
        } else {
            bufferAccumulator.receive(record, subpartitionId, dataType, isBroadcast);
        }
    }
@TestTemplate
    void testWriteRecordsToEmptyStorageTiers() {
        int numSubpartitions = 10;
        int bufferSize = 1024;
        Random random = new Random();

        TieredStorageProducerClient tieredStorageProducerClient =
                createTieredStorageProducerClient(numSubpartitions, Collections.emptyList());
        assertThatThrownBy(
                        () ->
                                tieredStorageProducerClient.write(
                                        generateRandomData(bufferSize, random),
                                        new TieredStorageSubpartitionId(0),
                                        Buffer.DataType.DATA_BUFFER,
                                        isBroadcast))
                .isInstanceOf(RuntimeException.class)
                .hasMessageContaining("Failed to choose a storage tier");
    }