public abstract void broadcastEmit(T record) throws IOException;
@TestTemplate
    void testBroadcastEmitRecord(@TempDir Path tempPath) throws Exception {
        final int numberOfSubpartitions = 4;
        final int bufferSize = 32;
        final int numValues = 8;
        final int serializationLength = 4;

        final ResultPartition partition = createResultPartition(bufferSize, numberOfSubpartitions);
        final RecordWriter<SerializationTestType> writer = createRecordWriter(partition);
        final RecordDeserializer<SerializationTestType> deserializer =
                new SpillingAdaptiveSpanningRecordDeserializer<>(
                        new String[] {tempPath.toString()});

        final ArrayDeque<SerializationTestType> serializedRecords = new ArrayDeque<>();
        final Iterable<SerializationTestType> records =
                Util.randomRecords(numValues, SerializationTestTypeFactory.INT);
        for (SerializationTestType record : records) {
            serializedRecords.add(record);
            writer.broadcastEmit(record);
        }

        final int numRequiredBuffers = numValues / (bufferSize / (4 + serializationLength));
        if (isBroadcastWriter) {
            assertThat(partition.getBufferPool().bestEffortGetNumOfUsedBuffers())
                    .isEqualTo(numRequiredBuffers);
        } else {
            assertThat(partition.getBufferPool().bestEffortGetNumOfUsedBuffers())
                    .isEqualTo(numRequiredBuffers * numberOfSubpartitions);
        }

        for (int i = 0; i < numberOfSubpartitions; i++) {
            assertThat(partition.getNumberOfQueuedBuffers(i)).isEqualTo(numRequiredBuffers);
            ResultSubpartitionView view =
                    partition.createSubpartitionView(
                            new ResultSubpartitionIndexSet(i), new NoOpBufferAvailablityListener());
            verifyDeserializationResults(
                    view, deserializer, serializedRecords.clone(), numRequiredBuffers, numValues);
        }
    }