@Override
                public DataStreamSink<?> consumeDataStream(
                        ProviderContext providerContext, DataStream<RowData> dataStream) {
                    final boolean objectReuse =
                            dataStream.getExecutionEnvironment().getConfig().isObjectReuseEnabled();
                    final ReducingUpsertSink<?, ?> sink =
                            new ReducingUpsertSink<>(
                                    kafkaSink,
                                    physicalDataType,
                                    keyProjection,
                                    flushMode,
                                    objectReuse
                                            ? createRowDataTypeSerializer(
                                                    context,
                                                    dataStream.getExecutionConfig())::copy
                                            : rowData -> rowData);
                    final DataStreamSink<RowData> end = dataStream.sinkTo(sink);
                    providerContext.generateUid(UPSERT_KAFKA_TRANSFORMATION).ifPresent(end::uid);
                    if (parallelism != null) {
                        end.setParallelism(parallelism);
                    }
                    return end;
                }
Parallelism of the physical Kafka producer. *
inlong-sort/sort-flink/sort-flink-v1.18/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicSink.java