[Feature] PIP-30: Run commit on the JobManager for unaware-bucket append tables

docs/generated/flink_connector_configuration.html

@@ -266,6 +266,12 @@
             <td>Boolean</td>
             <td>Allow sink committer and writer operator to be chained together</td>
         </tr>
+        <tr>
+            <td><h5>sink.coordinator-commit.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>If true, run the Paimon committer inside the Flink JobManager via an OperatorCoordinator. This decouples commit from any single TaskManager subtask so that region failover does not have to restart the whole pipeline. Only supports unaware-bucket append tables in streaming mode with checkpointing enabled; unsupported configurations fail during sink planning.</td>
+        </tr>
         <tr>
             <td><h5>sink.cross-partition.managed-memory</h5></td>
             <td style="word-wrap: break-word;">256 mb</td>

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java

@@ -486,6 +486,17 @@ public class FlinkConnectorOptions {
                             "Commit listener will be called after a successful commit. This option list custom commit "
                                     + "listener identifiers separated by comma.");
 
+    public static final ConfigOption<Boolean> SINK_COORDINATOR_COMMIT_ENABLED =
+            key("sink.coordinator-commit.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "If true, run the Paimon committer inside the Flink JobManager via an "
+                                    + "OperatorCoordinator. This decouples commit from any single TaskManager "
+                                    + "subtask so that region failover does not have to restart the whole pipeline. "
+                                    + "Only supports unaware-bucket append tables in streaming mode with "
+                                    + "checkpointing enabled; unsupported configurations fail during sink planning.");
+
     public static final ConfigOption<Boolean> SINK_WRITER_COORDINATOR_ENABLED =
             key("sink.writer-coordinator.enabled")
                     .booleanType()

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

@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.sink;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.flink.sink.coordinator.CommittableEvent;
+import org.apache.paimon.flink.sink.coordinator.CommittingWriteOperatorCoordinator;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.CommitMessageSerializer;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
+import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+
+/**
+ * Write operator that hands committables to a JM-side {@link CommittingWriteOperatorCoordinator}
+ * which performs the commit.
+ *
+ * <p>This operator is stateful: it keeps an independent operator state that buffers the
+ * per-checkpoint committables which have not yet been acknowledged by the coordinator, so they
+ * survive a global failover and can be replayed on restore.
+ */
+public class CoordinatorCommittingRowDataStoreWriteOperator
+        extends StatelessRowDataStoreWriteOperator {
+
+    private static final long serialVersionUID = 1L;
+
+    private static final Logger LOG =
+            LoggerFactory.getLogger(CoordinatorCommittingRowDataStoreWriteOperator.class);
+
+    @VisibleForTesting
+    static final String PENDING_COMMITTABLE_STATE_NAME = "pending_committable_state";
+
+    private final OperatorEventGateway operatorEventGateway;
+
+    /** Persisted buffer of committables not yet acknowledged by the coordinator. */
+    private transient ListState<Committable> pendingCommittableState;
+
+    /** In-memory view of {@link #pendingCommittableState}, grouped by checkpoint id. */
+    private transient NavigableMap<Long, List<Committable>> pendingCommittables;
+
+    private transient ListSerializer<Committable> serializer;
+
+    public CoordinatorCommittingRowDataStoreWriteOperator(
+            StreamOperatorParameters<Committable> parameters,
+            FileStoreTable table,
+            StoreSinkWrite.Provider storeSinkWriteProvider,
+            String initialCommitUser,
+            OperatorEventGateway operatorEventGateway) {
+        super(parameters, table, storeSinkWriteProvider, initialCommitUser);
+        this.operatorEventGateway = Preconditions.checkNotNull(operatorEventGateway);
+    }
+
+    @Override
+    public void initializeState(StateInitializationContext context) throws Exception {
+        super.initializeState(context);
+        serializer = CommittableEvent.createCommittableListSerializer();
+        pendingCommittableState =
+                new SimpleVersionedListState<>(
+                        context.getOperatorStateStore()
+                                .getListState(
+                                        new ListStateDescriptor<>(
+                                                PENDING_COMMITTABLE_STATE_NAME,
+                                                BytePrimitiveArraySerializer.INSTANCE)),
+                        new CommittableSerializer(new CommitMessageSerializer()));
+        pendingCommittables = new TreeMap<>();
+
+        if (context.isRestored()) {
+            // Always replay on restore, even if the buffer is empty: the coordinator relies on
+            // this signal to drive its own restore alignment.
+            Preconditions.checkState(context.getRestoredCheckpointId().isPresent());
+            long checkpointId = context.getRestoredCheckpointId().getAsLong();
+            List<Committable> restored = new ArrayList<>();
+            pendingCommittableState.get().forEach(restored::add);
+            LOG.info("Restore {} of checkpoint {} from state", restored, checkpointId);
+            pendingCommittableState.clear();
+            sendCommittableEventToCoordinator(checkpointId, true, restored);
+        }
+    }
+
+    @Override
+    public void snapshotState(StateSnapshotContext context) throws Exception {
+        super.snapshotState(context);
+        pendingCommittableState.clear();
+        for (List<Committable> committables : pendingCommittables.values()) {
+            pendingCommittableState.addAll(committables);
+        }
+    }
+
+    @Override
+    public void notifyCheckpointComplete(long checkpointId) throws Exception {
+        super.notifyCheckpointComplete(checkpointId);
+        // operator state already persisted these; we no longer need to replay them on the next
+        // restore
+        Map<Long, List<Committable>> completed = pendingCommittables.headMap(checkpointId, true);
+        completed.clear();
+    }
+
+    @Override
+    protected void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException {
+        List<Committable> committables = prepareCommit(waitCompaction, checkpointId);
+        // send the event regardless of whether the list is empty: the coordinator uses an event
+        // per (subtask, checkpoint) for alignment
+        sendCommittableEventToCoordinator(checkpointId, false, committables);
+        if (!committables.isEmpty()) {
+            pendingCommittables.put(checkpointId, committables);
+        }
+        // Still forward committables downstream even though the commit happens on the coordinator.
+        // The downstream is a DiscardingSink, but emitting keeps numRecordsOut observable and
+        // preserves the operator's IO metrics.
+        committables.forEach(committable -> output.collect(new StreamRecord<>(committable)));
+    }
+
+    private void sendCommittableEventToCoordinator(
+            long checkpointId, boolean isRestoring, List<Committable> committables)
+            throws IOException {
+        CommittableEvent event =
+                CommittableEvent.create(checkpointId, isRestoring, committables, serializer);
+        operatorEventGateway.sendEventToCoordinator(event);
+    }
+
+    @VisibleForTesting
+    NavigableMap<Long, List<Committable>> getPendingCommittables() {
+        return pendingCommittables;
+    }
+}

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

@@ -19,13 +19,15 @@
 package org.apache.paimon.flink.sink;
 
 import org.apache.paimon.CoreOptions.TagCreationMode;
+import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.flink.compact.changelog.ChangelogCompactCoordinateOperator;
 import org.apache.paimon.flink.compact.changelog.ChangelogCompactSortOperator;
 import org.apache.paimon.flink.compact.changelog.ChangelogCompactWorkerOperator;
 import org.apache.paimon.flink.compact.changelog.ChangelogTaskTypeInfo;
 import org.apache.paimon.manifest.ManifestCommittable;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.FileStoreTable;
 
 import org.apache.flink.api.common.RuntimeExecutionMode;
@@ -50,6 +52,7 @@
 import java.util.Queue;
 import java.util.Set;
 
+import static org.apache.paimon.CoreOptions.WRITE_ONLY;
 import static org.apache.paimon.CoreOptions.createCommitUser;
 import static org.apache.paimon.flink.FlinkConnectorOptions.END_INPUT_WATERMARK;
 import static org.apache.paimon.flink.FlinkConnectorOptions.PRECOMMIT_COMPACT;
@@ -187,15 +190,39 @@ public DataStream<Committable> doWrite(
 
     public DataStreamSink<?> doCommit(DataStream<Committable> written, String commitUser) {
         StreamExecutionEnvironment env = written.getExecutionEnvironment();
-        ReadableConfig conf = env.getConfiguration();
         CheckpointConfig checkpointConfig = env.getCheckpointConfig();
         boolean streamingCheckpointEnabled =
                 isStreaming(written) && checkpointConfig.isCheckpointingEnabled();
         if (streamingCheckpointEnabled) {
             assertStreamingConfiguration(env);
         }
 
+        if (coordinatorCommitEnabled()) {
+            return doCoordinatorCommit(written, checkpointConfig, streamingCheckpointEnabled);
+        }
+        return doOperatorCommit(written, commitUser, streamingCheckpointEnabled);
+    }
+
+    private DataStreamSink<?> doCoordinatorCommit(
+            DataStream<Committable> written,
+            CheckpointConfig checkpointConfig,
+            boolean streamingCheckpointEnabled) {
+        checkCoordinatorCommitPreconditions(table, checkpointConfig, streamingCheckpointEnabled);
+        // The commit runs inside the writer's OperatorCoordinator on the JobManager, so there
+        // is no global committer operator. Committables are still forwarded by the writer for
+        // observability and are discarded here.
+        return written.sinkTo(new PaimonDiscardingSink<>(table))
+                .name("end")
+                .setParallelism(written.getParallelism());
+    }
+
+    private DataStreamSink<?> doOperatorCommit(
+            DataStream<Committable> written,
+            String commitUser,
+            boolean streamingCheckpointEnabled) {
+        ReadableConfig conf = written.getExecutionEnvironment().getConfiguration();
         Options options = Options.fromMap(table.options());
+
         OneInputStreamOperatorFactory<Committable, Committable> committerOperator =
                 createCommitterOperatorFactory(
                         streamingCheckpointEnabled, commitUser, options.get(END_INPUT_WATERMARK));
@@ -310,6 +337,74 @@ protected abstract OneInputStreamOperatorFactory<T, Committable> createWriteOper
 
     protected abstract CommittableStateManager<ManifestCommittable> createCommittableStateManager();
 
+    /**
+     * Whether the writer commits through its {@link
+     * org.apache.flink.runtime.operators.coordination.OperatorCoordinator} on the JobManager. When
+     * {@code true}, {@link #doCommit} skips the global committer operator. Sinks that wire a
+     * coordinated writer factory override this.
+     */
+    protected boolean coordinatorCommitEnabled() {
+        return false;
+    }
+
+    /**
+     * Fails fast when coordinator commit is enabled together with a table property or runtime
+     * setting it cannot honor yet. Enabling coordinator commit is an explicit user choice, so a
+     * conflicting configuration is reported instead of silently falling back to the global
+     * committer.
+     */
+    @VisibleForTesting
+    static void checkCoordinatorCommitPreconditions(
+            FileStoreTable table,
+            CheckpointConfig checkpointConfig,
+            boolean streamingCheckpointEnabled) {
+        Options options = Options.fromMap(table.options());
+
+        // Region failover only benefits streaming jobs. A batch job persists its shuffle data and
+        // has no region-failover concern, so coordinator commit brings no benefit and batch is not
+        // supported. The commit is driven by checkpoint completion, so checkpointing must be on.
+        checkArgument(
+                streamingCheckpointEnabled,
+                "Could not enable coordinator commit because it requires streaming mode with checkpointing enabled.");
+
+        // The checks below reject configurations that introduce an all-to-all shuffle. Such a
+        // shuffle places every operator into a single failover region, which defeats the region
+        // failover that coordinator commit is meant to enable.
+        checkArgument(
+                table.primaryKeys().isEmpty(),
+                "Could not enable coordinator commit because only append tables are supported, but the table has primary keys.");
+        checkArgument(
+                table.bucketMode() == BucketMode.BUCKET_UNAWARE,
+                "Could not enable coordinator commit because bucket mode is "
+                        + table.bucketMode()
+                        + ", only unaware bucket is supported.");
+        checkArgument(
+                table.coreOptions().writeOnly(),
+                "Could not enable coordinator commit because it requires "
+                        + WRITE_ONLY.key()
+                        + " = true.");
+        checkArgument(
+                !options.get(PRECOMMIT_COMPACT),
+                "Could not enable coordinator commit because it requires "
+                        + PRECOMMIT_COMPACT.key()
+                        + " = false.");
+
+        // The OperatorCoordinator cannot tell a savepoint from a normal checkpoint.
+        // TODO support savepoint auto-tag.
+        checkArgument(
+                !options.get(SINK_AUTO_TAG_FOR_SAVEPOINT),
+                "Could not enable coordinator commit because "
+                        + SINK_AUTO_TAG_FOR_SAVEPOINT.key()
+                        + " is enabled, which is not supported yet.");
+
+        // TODO concurrent checkpoints are not supported yet.
+        checkArgument(
+                checkpointConfig.getMaxConcurrentCheckpoints() == 1,
+                "Could not enable coordinator commit because execution.checkpointing.max-concurrent-checkpoints is "
+                        + checkpointConfig.getMaxConcurrentCheckpoints()
+                        + ", which is not supported yet.");
+    }
+
     public static boolean isStreaming(DataStream<?> input) {
         return isStreaming(input.getExecutionEnvironment());
     }

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

@@ -111,7 +111,7 @@ public void close() throws Exception {
         }
     }
 
-    private void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException {
+    protected void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException {
         prepareCommit(waitCompaction, checkpointId)
                 .forEach(committable -> output.collect(new StreamRecord<>(committable)));
     }