fix sink.parallelism not work

tongcheng-elong · May 6, 2024 · e6ad289 · e6ad289
1 parent 07df344
commit e6ad289
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -162,7 +162,7 @@ public SingleOutputStreamOperator<Committable> doWrite(
                 conf.get(ExecutionOptions.RUNTIME_MODE) == RuntimeExecutionMode.STREAMING;
 
         // partitioner parallelism
-        Integer calParallelism = input.getParallelism();
+        Integer calParallelism = parallelism == null ? input.getParallelism() : parallelism;
         if (table.options().get(FlinkConnectorOptions.SINK_PARALLELISM.key()) == null) {
             // batch and with writeManifestCache
             if (!isStreaming && table.coreOptions().writeManifestCache().getMebiBytes() > 0) {