[flink] Support end input in coordinator commit

docs/generated/flink_connector_configuration.html

@@ -276,7 +276,7 @@
             <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>
+            <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. Streaming mode requires checkpointing enabled; batch mode commits after all writers reach end input. Unsupported configurations fail during sink planning.</td>
         </tr>
         <tr>
             <td><h5>sink.cross-partition.managed-memory</h5></td>

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

@@ -504,8 +504,9 @@ public class FlinkConnectorOptions {
                             "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.");
+                                    + "Only supports unaware-bucket append tables. Streaming mode requires "
+                                    + "checkpointing enabled; batch mode commits after all writers reach end input. "
+                                    + "Unsupported configurations fail during sink planning.");
 
     public static final ConfigOption<Boolean> SINK_WRITER_COORDINATOR_ENABLED =
             key("sink.writer-coordinator.enabled")

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

@@ -65,6 +65,7 @@ public class CoordinatorCommittingRowDataStoreWriteOperator
 
     private static final Logger LOG =
             LoggerFactory.getLogger(CoordinatorCommittingRowDataStoreWriteOperator.class);
+    private static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
 
     @VisibleForTesting
     static final String PENDING_COMMITTABLE_STATE_NAME = "pending_committable_state";
@@ -134,6 +135,11 @@ public void initializeState(StateInitializationContext context) throws Exception
 
             List<CheckpointCommittables> restored = new ArrayList<>();
             for (CheckpointCommittables entry : pendingCommittableState.get()) {
+                // End input is newer than every ordinary restored checkpoint and must survive
+                // subsequent snapshots even if Flink does not call endInput again after restore.
+                if (entry.checkpointId() == END_INPUT_CHECKPOINT_ID) {
+                    pendingCommittables.put(entry.checkpointId(), entry);
+                }
                 restored.add(entry);
             }
             pendingCommittableState.clear();
@@ -172,6 +178,18 @@ protected void emitCommittables(boolean waitCompaction, long checkpointId) throw
         CheckpointCommittables entry =
                 new CheckpointCommittables(
                         checkpointId, committables, currentWatermark, currentIdle);
+        if (checkpointId == END_INPUT_CHECKPOINT_ID) {
+            CheckpointCommittables previous = pendingCommittables.get(checkpointId);
+            if (previous != null) {
+                // A restored writer may receive endInput again. Preserve the previously persisted
+                // final committables and send one authoritative entry to the coordinator.
+                List<Committable> merged = new ArrayList<>(previous.committables());
+                merged.addAll(entry.committables());
+                entry =
+                        new CheckpointCommittables(
+                                checkpointId, merged, currentWatermark, currentIdle);
+            }
+        }
         // Emit an event per (subtask, checkpoint) regardless of whether committables is empty.
         operatorEventGateway.sendEventToCoordinator(
                 CommittableEvent.create(checkpointId, entry, eventSerializer));

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

@@ -53,6 +53,7 @@
 import java.util.Queue;
 import java.util.Set;
 
+import static org.apache.paimon.CoreOptions.TAG_AUTOMATIC_CREATION;
 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;
@@ -135,6 +136,9 @@ public DataStream<Committable> doWrite(
             DataStream<T> input, String commitUser, @Nullable Integer parallelism) {
         StreamExecutionEnvironment env = input.getExecutionEnvironment();
         boolean isStreaming = isStreaming(input);
+        boolean streamingCheckpointEnabled =
+                isStreaming && env.getCheckpointConfig().isCheckpointingEnabled();
+        Options options = Options.fromMap(table.options());
 
         boolean writeOnly = table.coreOptions().writeOnly();
         StoreSinkWrite.Provider writeProvider =
@@ -152,15 +156,17 @@ public DataStream<Committable> doWrite(
                 input.transform(
                         (writeOnly ? WRITER_WRITE_ONLY_NAME : WRITER_NAME) + " : " + table.name(),
                         new CommittableTypeInfo(),
-                        createWriteOperatorFactory(writeProvider, commitUser));
+                        createWriteOperatorFactory(
+                                writeProvider,
+                                commitUser,
+                                streamingCheckpointEnabled,
+                                options.get(END_INPUT_WATERMARK)));
         if (parallelism == null) {
             forwardParallelism(written, input);
         } else {
             written.setParallelism(parallelism);
         }
 
-        Options options = Options.fromMap(table.options());
-
         String uidSuffix = options.get(SINK_OPERATOR_UID_SUFFIX);
         if (options.get(SINK_OPERATOR_UID_SUFFIX) != null) {
             written = written.uid(generateCustomUid(WRITER_NAME, table.name(), uidSuffix));
@@ -202,23 +208,27 @@ public DataStream<Committable> doWrite(
     public DataStreamSink<?> doCommit(DataStream<Committable> written, String commitUser) {
         StreamExecutionEnvironment env = written.getExecutionEnvironment();
         CheckpointConfig checkpointConfig = env.getCheckpointConfig();
+        boolean isStreaming = isStreaming(written);
         boolean streamingCheckpointEnabled =
-                isStreaming(written) && checkpointConfig.isCheckpointingEnabled();
+                isStreaming && checkpointConfig.isCheckpointingEnabled();
         if (streamingCheckpointEnabled) {
             assertStreamingConfiguration(env);
         }
 
         if (coordinatorCommitEnabled()) {
-            return doCoordinatorCommit(written, checkpointConfig, streamingCheckpointEnabled);
+            return doCoordinatorCommit(
+                    written, checkpointConfig, isStreaming, streamingCheckpointEnabled);
         }
         return doOperatorCommit(written, commitUser, streamingCheckpointEnabled);
     }
 
     private DataStreamSink<?> doCoordinatorCommit(
             DataStream<Committable> written,
             CheckpointConfig checkpointConfig,
+            boolean isStreaming,
             boolean streamingCheckpointEnabled) {
-        checkCoordinatorCommitPreconditions(table, checkpointConfig, streamingCheckpointEnabled);
+        checkCoordinatorCommitPreconditions(
+                table, checkpointConfig, isStreaming, 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.
@@ -341,6 +351,14 @@ public static void assertBatchAdaptiveParallelism(
                 endInputWatermark);
     }
 
+    protected OneInputStreamOperatorFactory<T, Committable> createWriteOperatorFactory(
+            StoreSinkWrite.Provider writeProvider,
+            String commitUser,
+            boolean streamingCheckpointEnabled,
+            @Nullable Long endInputWatermark) {
+        return createWriteOperatorFactory(writeProvider, commitUser);
+    }
+
     protected abstract OneInputStreamOperatorFactory<T, Committable> createWriteOperatorFactory(
             StoreSinkWrite.Provider writeProvider, String commitUser);
 
@@ -368,15 +386,15 @@ protected boolean coordinatorCommitEnabled() {
     static void checkCoordinatorCommitPreconditions(
             FileStoreTable table,
             CheckpointConfig checkpointConfig,
+            boolean isStreaming,
             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.
+        // Streaming commits are driven by checkpoint completion. Batch jobs commit when every
+        // writer reaches endInput, so they do not require checkpointing.
         checkArgument(
-                streamingCheckpointEnabled,
-                "Could not enable coordinator commit because it requires streaming mode with checkpointing enabled.");
+                !isStreaming || streamingCheckpointEnabled,
+                "Could not enable coordinator commit in streaming mode without checkpointing.");
 
         // 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
@@ -407,6 +425,12 @@ static void checkCoordinatorCommitPreconditions(
                 "Could not enable coordinator commit because "
                         + SINK_AUTO_TAG_FOR_SAVEPOINT.key()
                         + " is enabled, which is not supported yet.");
+        // TODO support batch tag create.
+        checkArgument(
+                table.coreOptions().tagCreationMode() != TagCreationMode.BATCH,
+                "Could not enable coordinator commit because "
+                        + TAG_AUTOMATIC_CREATION.key()
+                        + " = batch is not supported yet.");
 
         // TODO concurrent checkpoints are not supported yet.
         checkArgument(

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

@@ -33,6 +33,8 @@
 import org.apache.flink.streaming.api.operators.StreamOperator;
 import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
 
+import javax.annotation.Nullable;
+
 import java.util.Map;
 
 /** An {@link AppendTableSink} which handles {@link InternalRow}. */
@@ -48,16 +50,24 @@ public RowAppendTableSink(
     @Override
     protected OneInputStreamOperatorFactory<InternalRow, Committable> createWriteOperatorFactory(
             StoreSinkWrite.Provider writeProvider, String commitUser) {
+        return createWriteOperatorFactory(writeProvider, commitUser, true, null);
+    }
+
+    @Override
+    protected OneInputStreamOperatorFactory<InternalRow, Committable> createWriteOperatorFactory(
+            StoreSinkWrite.Provider writeProvider,
+            String commitUser,
+            boolean streamingCheckpointEnabled,
+            @Nullable Long endInputWatermark) {
         if (coordinatorCommitEnabled()) {
             return createCoordinatorCommittingRowWriteOperatorFactory(
                     table,
                     writeProvider,
                     commitUser,
-                    // checkpointing on by default for the JM-side committer; bounded sources will
-                    // be handled by end-input support in a follow-up PR
-                    true,
+                    streamingCheckpointEnabled,
                     true,
-                    createCommitterFactory());
+                    createCommitterFactory(),
+                    endInputWatermark);
         }
         return createNoStateRowWriteOperatorFactory(table, writeProvider, commitUser);
     }
@@ -84,14 +94,16 @@ protected CommittableStateManager<ManifestCommittable> createCommittableStateMan
                     String commitUser,
                     boolean streamingCheckpointEnabled,
                     boolean failoverAfterRecovery,
-                    Committer.Factory<Committable, ManifestCommittable> committerFactory) {
+                    Committer.Factory<Committable, ManifestCommittable> committerFactory,
+                    @Nullable Long endInputWatermark) {
         return new CoordinatorCommittingFactory(
                 table,
                 writeProvider,
                 commitUser,
                 streamingCheckpointEnabled,
                 failoverAfterRecovery,
-                committerFactory);
+                committerFactory,
+                endInputWatermark);
     }
 
     private static class CoordinatorCommittingFactory extends RowDataStoreWriteOperator.Factory
@@ -102,18 +114,21 @@ private static class CoordinatorCommittingFactory extends RowDataStoreWriteOpera
         private final boolean streamingCheckpointEnabled;
         private final boolean failoverAfterRecovery;
         private final Committer.Factory<Committable, ManifestCommittable> committerFactory;
+        private final Long endInputWatermark;
 
         CoordinatorCommittingFactory(
                 FileStoreTable table,
                 StoreSinkWrite.Provider storeSinkWriteProvider,
                 String initialCommitUser,
                 boolean streamingCheckpointEnabled,
                 boolean failoverAfterRecovery,
-                Committer.Factory<Committable, ManifestCommittable> committerFactory) {
+                Committer.Factory<Committable, ManifestCommittable> committerFactory,
+                @Nullable Long endInputWatermark) {
             super(table, storeSinkWriteProvider, initialCommitUser);
             this.streamingCheckpointEnabled = streamingCheckpointEnabled;
             this.failoverAfterRecovery = failoverAfterRecovery;
             this.committerFactory = committerFactory;
+            this.endInputWatermark = endInputWatermark;
         }
 
         @Override
@@ -124,7 +139,8 @@ public OperatorCoordinator.Provider getCoordinatorProvider(
                     committerFactory,
                     streamingCheckpointEnabled,
                     initialCommitUser,
-                    failoverAfterRecovery);
+                    failoverAfterRecovery,
+                    endInputWatermark);
         }
 
         @Override