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 testWriteRecords() throws IOException {
        int numSubpartitions = 10;
        int numToWriteRecords = 20;
        int bufferSize = 1024;
        Random random = new Random();

        AtomicInteger numReceivedBuffers = new AtomicInteger(0);
        AtomicInteger numReceivedBytes = new AtomicInteger(0);
        AtomicInteger numReceivedBuffersInTier1 = new AtomicInteger(0);
        AtomicInteger numReceivedBuffersInTier2 = new AtomicInteger(0);

        TestingTierProducerAgent tierProducerAgent1 =
                new TestingTierProducerAgent.Builder()
                        .setTryStartSegmentSupplier(
                                ((subpartitionId, integer) -> numReceivedBuffersInTier1.get() < 1))
                        .setTryWriterFunction(
                                ((subpartitionId, buffer) -> {
                                    boolean isSuccess = numReceivedBuffersInTier1.get() % 2 == 0;
                                    if (isSuccess) {
                                        numReceivedBuffers.incrementAndGet();
                                        numReceivedBuffersInTier1.incrementAndGet();
                                        numReceivedBytes.set(
                                                numReceivedBytes.get() + buffer.readableBytes());
                                    }
                                    return isSuccess;
                                }))
                        .build();
        TestingTierProducerAgent tierProducerAgent2 =
                new TestingTierProducerAgent.Builder()
                        .setTryWriterFunction(
                                ((subpartitionId, buffer) -> {
                                    numReceivedBuffers.incrementAndGet();
                                    numReceivedBuffersInTier2.incrementAndGet();
                                    numReceivedBytes.set(
                                            numReceivedBytes.get() + buffer.readableBytes());
                                    return true;
                                }))
                        .build();
        List<TierProducerAgent> tierProducerAgents = new ArrayList<>();
        tierProducerAgents.add(tierProducerAgent1);
        tierProducerAgents.add(tierProducerAgent2);

        TieredStorageProducerClient tieredStorageProducerClient =
                createTieredStorageProducerClient(numSubpartitions, tierProducerAgents);
        TieredStorageSubpartitionId subpartitionId = new TieredStorageSubpartitionId(0);

        for (int i = 0; i < numToWriteRecords; i++) {
            tieredStorageProducerClient.write(
                    generateRandomData(bufferSize, random),
                    subpartitionId,
                    Buffer.DataType.DATA_BUFFER,
                    isBroadcast);
        }

        int numExpectedBytes =
                isBroadcast
                        ? numSubpartitions * numToWriteRecords * bufferSize
                        : numToWriteRecords * bufferSize;
        assertThat(numReceivedBuffersInTier1.get()).isEqualTo(1);
        assertThat(numReceivedBuffers.get())
                .isEqualTo(numReceivedBuffersInTier1.get() + numReceivedBuffersInTier2.get());
        assertThat(numReceivedBytes.get()).isEqualTo(numExpectedBytes);
    }