@VisibleForTesting
    public static <OUT>
            RecordWriterDelegate<SerializationDelegate<StreamRecord<OUT>>>
                    createRecordWriterDelegate(
                            StreamConfig configuration, Environment environment) {
        List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> recordWrites =
                createRecordWriters(configuration, environment);
        if (recordWrites.size() == 1) {
            return new SingleRecordWriter<>(recordWrites.get(0));
        } else if (recordWrites.size() == 0) {
            return new NonRecordWriter<>();
        } else {
            return new MultipleRecordWriters<>(recordWrites);
        }
    }
@Test
    void testForwardPartitionerIsConvertedToRebalanceOnParallelismChanges() throws Exception {
        StreamTaskMailboxTestHarnessBuilder<Integer> builder =
                new StreamTaskMailboxTestHarnessBuilder<>(
                                OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
                        .addInput(BasicTypeInfo.INT_TYPE_INFO)
                        .setOutputPartitioner(new ForwardPartitioner<>())
                        .setupOutputForSingletonOperatorChain(
                                new TestBoundedOneInputStreamOperator());

        try (StreamTaskMailboxTestHarness<Integer> harness = builder.build()) {

            RecordWriterDelegate<SerializationDelegate<StreamRecord<Object>>> recordWriterDelegate =
                    harness.streamTask.createRecordWriterDelegate(
                            harness.streamTask.configuration, harness.streamMockEnvironment);
            // Prerequisite: We are using the ForwardPartitioner
            assertThat(
                            ((ChannelSelectorRecordWriter)
                                            ((SingleRecordWriter) recordWriterDelegate)
                                                    .getRecordWriter(0))
                                    .getChannelSelector())
                    .isInstanceOf(ForwardPartitioner.class);

            // Simulate changed downstream task parallelism (1->2)
            List<ResultPartitionWriter> newOutputs = new ArrayList<>();
            newOutputs.add(
                    new MockResultPartitionWriter() {
                        @Override
                        public int getNumberOfSubpartitions() {
                            return 2;
                        }
                    });
            harness.streamMockEnvironment.setOutputs(newOutputs);

            // Re-create outputs
            recordWriterDelegate =
                    harness.streamTask.createRecordWriterDelegate(
                            harness.streamTask.configuration, harness.streamMockEnvironment);
            // We should now have a RebalancePartitioner to distribute the load
            // for the non-matching downstream parallelism
            assertThat(
                            ((ChannelSelectorRecordWriter)
                                            ((SingleRecordWriter) recordWriterDelegate)
                                                    .getRecordWriter(0))
                                    .getChannelSelector())
                    .isInstanceOf(RebalancePartitioner.class);
        }
    }