[flink] Check parallelism of committer operator while runtime.

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java

@@ -127,6 +127,7 @@ public DataStreamSink<?> sinkFrom(
                                 typeInfo,
                                 new CommitterOperator<>(
                                         true,
+                                        false,
                                         commitUser,
                                         createCommitterFactory(),
                                         createCommittableStateManager()))

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java

@@ -18,6 +18,8 @@
 
 package org.apache.paimon.flink.sink;
 
+import org.apache.paimon.utils.Preconditions;
+
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
@@ -55,6 +57,9 @@ public class CommitterOperator<CommitT, GlobalCommitT> extends AbstractStreamOpe
      */
     private final boolean streamingCheckpointEnabled;
 
+    /** Whether to check the parallelism while runtime. */
+    private final boolean forceSingleParallelism;
+
     /**
      * This commitUser is valid only for new jobs. After the job starts, this commitUser will be
      * recorded into the states of write and commit operators. When the job restarts, commitUser
@@ -83,10 +88,12 @@ public class CommitterOperator<CommitT, GlobalCommitT> extends AbstractStreamOpe
 
     public CommitterOperator(
             boolean streamingCheckpointEnabled,
+            boolean forceSingleParallelism,
             String initialCommitUser,
             Committer.Factory<CommitT, GlobalCommitT> committerFactory,
             CommittableStateManager<GlobalCommitT> committableStateManager) {
         this.streamingCheckpointEnabled = streamingCheckpointEnabled;
+        this.forceSingleParallelism = forceSingleParallelism;
         this.initialCommitUser = initialCommitUser;
         this.committablesPerCheckpoint = new TreeMap<>();
         this.committerFactory = checkNotNull(committerFactory);
@@ -98,6 +105,10 @@ public CommitterOperator(
     public void initializeState(StateInitializationContext context) throws Exception {
         super.initializeState(context);
 
+        Preconditions.checkArgument(
+                !forceSingleParallelism || getRuntimeContext().getNumberOfParallelSubtasks() == 1,
+                "Committer Operator parallelism in paimon MUST be one.");
+
         this.currentWatermark = Long.MIN_VALUE;
         this.endInput = false;
         // each job can only have one user name and this name must be consistent across restarts

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java

@@ -228,6 +228,7 @@ protected DataStreamSink<?> doCommit(DataStream<Committable> written, String com
         OneInputStreamOperator<Committable, Committable> committerOperator =
                 new CommitterOperator<>(
                         streamingCheckpointEnabled,
+                        true,
                         commitUser,
                         createCommitterFactory(streamingCheckpointEnabled),
                         createCommittableStateManager());

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesCompactorSink.java

@@ -128,6 +128,7 @@ protected DataStreamSink<?> doCommit(
                                 new MultiTableCommittableTypeInfo(),
                                 new CommitterOperator<>(
                                         streamingCheckpointEnabled,
+                                        false,
                                         commitUser,
                                         createCommitterFactory(),
                                         createCommittableStateManager()))

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java

@@ -48,6 +48,7 @@
 import org.apache.flink.metrics.groups.OperatorMetricGroup;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.api.watermark.Watermark;
@@ -651,6 +652,19 @@ public void testCommitMetrics() throws Exception {
         write.close();
     }
 
+    @Test
+    public void testParallelism() throws Exception {
+        FileStoreTable table = createFileStoreTable();
+        String commitUser = UUID.randomUUID().toString();
+        OneInputStreamOperator<Committable, Committable> operator =
+                createCommitterOperator(table, commitUser, new NoopCommittableStateManager());
+        try (OneInputStreamOperatorTestHarness<Committable, Committable> testHarness =
+                createTestHarness(operator, 10, 10, 3)) {
+            Assertions.assertThatCode(testHarness::open)
+                    .hasMessage("Committer Operator parallelism in paimon MUST be one.");
+        }
+    }
+
     // ------------------------------------------------------------------------
     //  Test utils
     // ------------------------------------------------------------------------
@@ -682,10 +696,25 @@ private OneInputStreamOperatorTestHarness<Committable, Committable> createLossyT
 
     private OneInputStreamOperatorTestHarness<Committable, Committable> createTestHarness(
             OneInputStreamOperator<Committable, Committable> operator) throws Exception {
+        return createTestHarness(operator, 1, 1, 0);
+    }
+
+    private OneInputStreamOperatorTestHarness<Committable, Committable> createTestHarness(
+            OneInputStreamOperator<Committable, Committable> operator,
+            int maxParallelism,
+            int parallelism,
+            int subTaskIndex)
+            throws Exception {
         TypeSerializer<Committable> serializer =
                 new CommittableTypeInfo().createSerializer(new ExecutionConfig());
         OneInputStreamOperatorTestHarness<Committable, Committable> harness =
-                new OneInputStreamOperatorTestHarness<>(operator, serializer);
+                new OneInputStreamOperatorTestHarness<>(
+                        operator,
+                        maxParallelism,
+                        parallelism,
+                        subTaskIndex,
+                        serializer,
+                        new OperatorID());
         harness.setup(serializer);
         return harness;
     }
@@ -695,6 +724,7 @@ protected OneInputStreamOperator<Committable, Committable> createCommitterOperat
             String commitUser,
             CommittableStateManager<ManifestCommittable> committableStateManager) {
         return new CommitterOperator<>(
+                true,
                 true,
                 commitUser == null ? initialCommitUser : commitUser,
                 (user, metricGroup) ->
@@ -710,6 +740,7 @@ protected OneInputStreamOperator<Committable, Committable> createCommitterOperat
             CommittableStateManager<ManifestCommittable> committableStateManager,
             ThrowingConsumer<StateInitializationContext, Exception> initializeFunction) {
         return new CommitterOperator<Committable, ManifestCommittable>(
+                true,
                 true,
                 commitUser == null ? initialCommitUser : commitUser,
                 (user, metricGroup) ->

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/StoreMultiCommitterTest.java

@@ -636,6 +636,7 @@ public void testCommitMetrics() throws Exception {
         CommitterOperator<MultiTableCommittable, WrappedManifestCommittable> operator =
                 new CommitterOperator<>(
                         true,
+                        false,
                         initialCommitUser,
                         (user, metricGroup) ->
                                 new StoreMultiCommitter(
@@ -652,6 +653,7 @@ public void testCommitMetrics() throws Exception {
         CommitterOperator<MultiTableCommittable, WrappedManifestCommittable> operator =
                 new CommitterOperator<>(
                         true,
+                        false,
                         initialCommitUser,
                         (user, metricGroup) ->
                                 new StoreMultiCommitter(