Fix data loss when recovering incomplete END_INPUT committables

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

@@ -177,6 +177,8 @@ public DataStreamSink<?> sinkFrom(
 
     protected CommittableStateManager<WrappedManifestCommittable> createCommittableStateManager() {
         return new RestoreAndFailCommittableStateManager<>(
-                WrappedManifestCommittableSerializer::new, true);
+                WrappedManifestCommittableSerializer::new,
+                true,
+                StoreMultiCommitter.END_INPUT_HANDLER);
     }
 }

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

@@ -208,6 +208,7 @@ protected DataStreamSink<?> doCommit(
     protected CommittableStateManager<WrappedManifestCommittable> createCommittableStateManager() {
         return new RestoreAndFailCommittableStateManager<>(
                 WrappedManifestCommittableSerializer::new,
-                options.get(PARTITION_MARK_DONE_RECOVER_FROM_STATE));
+                options.get(PARTITION_MARK_DONE_RECOVER_FROM_STATE),
+                StoreMultiCommitter.END_INPUT_HANDLER);
     }
 }

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

@@ -27,8 +27,13 @@
  */
 public interface CommittableStateManager<GlobalCommitT> extends Serializable {
 
-    void initializeState(Committer.Context context, Committer<?, GlobalCommitT> committer)
-            throws Exception;
+    /**
+     * Initializes the state and returns restored committables which must remain pending in the
+     * operator.
+     */
+    List<GlobalCommitT> initializeState(
+            Committer.Context context, Committer<?, GlobalCommitT> committer) throws Exception;
 
-    void snapshotState(List<GlobalCommitT> committables) throws Exception;
+    /** Snapshots pending committables together with the complete end-input state. */
+    void snapshotState(List<GlobalCommitT> committables, boolean completeEndInput) throws Exception;
 }

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

@@ -47,7 +47,7 @@ public class CommitterOperator<CommitT, GlobalCommitT> extends AbstractStreamOpe
         implements OneInputStreamOperator<CommitT, CommitT>, BoundedOneInput {
 
     private static final long serialVersionUID = 1L;
-    private static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
+    static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
 
     /** Record all the inputs until commit. */
     private final Deque<CommitT> inputs = new ArrayDeque<>();
@@ -85,7 +85,7 @@ public class CommitterOperator<CommitT, GlobalCommitT> extends AbstractStreamOpe
 
     private transient long currentWatermark;
 
-    private transient boolean endInput;
+    private transient boolean completeEndInput;
 
     private transient String commitUser;
 
@@ -123,7 +123,7 @@ public void initializeState(StateInitializationContext context) throws Exception
                 "Committer Operator parallelism in paimon MUST be one.");
 
         this.currentWatermark = Long.MIN_VALUE;
-        this.endInput = false;
+        this.completeEndInput = false;
         // each job can only have one user name and this name must be consistent across restarts
         // we cannot use job id as commit user name here because user may change job id by creating
         // a savepoint, stop the job and then resume from savepoint
@@ -149,7 +149,14 @@ public void initializeState(StateInitializationContext context) throws Exception
                                 .getSpillingDirectoriesPaths());
         committer = committerFactory.create(committerContext);
 
-        committableStateManager.initializeState(committerContext, committer);
+        List<GlobalCommitT> pendingEndInputCommittables =
+                committableStateManager.initializeState(committerContext, committer);
+        for (GlobalCommitT committable : pendingEndInputCommittables) {
+            Preconditions.checkState(
+                    !committablesPerCheckpoint.containsKey(END_INPUT_CHECKPOINT_ID),
+                    "State manager returned multiple pending end-input committables.");
+            committablesPerCheckpoint.put(END_INPUT_CHECKPOINT_ID, committable);
+        }
     }
 
     @Override
@@ -170,7 +177,8 @@ public void snapshotState(StateSnapshotContext context) throws Exception {
         super.snapshotState(context);
         pollInputs();
         committer.snapshotState();
-        committableStateManager.snapshotState(committables(committablesPerCheckpoint));
+        committableStateManager.snapshotState(
+                committables(committablesPerCheckpoint), completeEndInput);
     }
 
     private List<GlobalCommitT> committables(NavigableMap<Long, GlobalCommitT> map) {
@@ -179,23 +187,24 @@ private List<GlobalCommitT> committables(NavigableMap<Long, GlobalCommitT> map)
 
     @Override
     public void endInput() throws Exception {
-        endInput = true;
         if (endInputWatermark != null) {
             currentWatermark = endInputWatermark;
         }
 
+        pollInputs();
+        completeEndInput = true;
+
         if (streamingCheckpointEnabled) {
             return;
         }
 
-        pollInputs();
         commitUpToCheckpoint(END_INPUT_CHECKPOINT_ID);
     }
 
     @Override
     public void notifyCheckpointComplete(long checkpointId) throws Exception {
         super.notifyCheckpointComplete(checkpointId);
-        commitUpToCheckpoint(endInput ? END_INPUT_CHECKPOINT_ID : checkpointId);
+        commitUpToCheckpoint(completeEndInput ? END_INPUT_CHECKPOINT_ID : checkpointId);
     }
 
     private void commitUpToCheckpoint(long checkpointId) throws Exception {

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

@@ -0,0 +1,34 @@
+/*
+ * 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 java.io.Serializable;
+
+/** Operations required to keep incomplete end-input committables pending during recovery. */
+public interface EndInputCommittableHandler<GlobalCommitT> extends Serializable {
+
+    /** Returns whether the committable belongs to end input. */
+    boolean isEndInput(GlobalCommitT committable);
+
+    /**
+     * Merges two restored end-input committables without rebuilding them or refreshing watermark
+     * metadata.
+     */
+    GlobalCommitT merge(GlobalCommitT target, GlobalCommitT source);
+}

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

@@ -66,7 +66,8 @@ protected CommittableStateManager<ManifestCommittable> createCommittableStateMan
         Options options = table.coreOptions().toConfiguration();
         return new RestoreAndFailCommittableStateManager<>(
                 ManifestCommittableSerializer::new,
-                options.get(PARTITION_MARK_DONE_RECOVER_FROM_STATE));
+                options.get(PARTITION_MARK_DONE_RECOVER_FROM_STATE),
+                StoreCommitter.END_INPUT_HANDLER);
     }
 
     protected static OneInputStreamOperatorFactory<InternalRow, Committable>
@@ -89,6 +90,7 @@ public StreamOperator createStreamOperator(StreamOperatorParameters parameters)
         Options options = table.coreOptions().toConfiguration();
         return new RestoreCommittableStateManager<>(
                 ManifestCommittableSerializer::new,
-                options.get(PARTITION_MARK_DONE_RECOVER_FROM_STATE));
+                options.get(PARTITION_MARK_DONE_RECOVER_FROM_STATE),
+                StoreCommitter.END_INPUT_HANDLER);
     }
 }

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

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.manifest.ManifestCommittable;
 
+import java.util.Collections;
 import java.util.List;
 
 /**
@@ -32,14 +33,15 @@
 public class NoopCommittableStateManager implements CommittableStateManager<ManifestCommittable> {
 
     @Override
-    public void initializeState(
+    public List<ManifestCommittable> initializeState(
             Committer.Context context, Committer<?, ManifestCommittable> committer)
             throws Exception {
-        // nothing to do
+        return Collections.emptyList();
     }
 
     @Override
-    public void snapshotState(List<ManifestCommittable> committables) throws Exception {
+    public void snapshotState(List<ManifestCommittable> committables, boolean completeEndInput)
+            throws Exception {
         // nothing to do
     }
 }

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

@@ -41,8 +41,9 @@ public class RestoreAndFailCommittableStateManager<GlobalCommitT>
 
     public RestoreAndFailCommittableStateManager(
             SerializableSupplier<VersionedSerializer<GlobalCommitT>> committableSerializer,
-            boolean partitionMarkDoneRecoverFromState) {
-        super(committableSerializer, partitionMarkDoneRecoverFromState);
+            boolean partitionMarkDoneRecoverFromState,
+            EndInputCommittableHandler<GlobalCommitT> endInputHandler) {
+        super(committableSerializer, partitionMarkDoneRecoverFromState, endInputHandler);
     }
 
     @Override

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

@@ -25,16 +25,19 @@
 
 import org.apache.flink.api.common.state.ListState;
 import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.BooleanSerializer;
 import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
 import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
 /**
  * A {@link CommittableStateManager} which stores uncommitted {@link ManifestCommittable}s in state.
  *
- * <p>When the job restarts, these {@link ManifestCommittable}s will be restored and committed.
+ * <p>When the job restarts, regular checkpoint committables are restored and committed. An
+ * incomplete END_INPUT committable remains pending until the operator receives complete end input.
  */
 public class RestoreCommittableStateManager<GlobalCommitT>
         implements CommittableStateManager<GlobalCommitT> {
@@ -46,19 +49,41 @@ public class RestoreCommittableStateManager<GlobalCommitT>
 
     private final boolean partitionMarkDoneRecoverFromState;
 
+    private final EndInputCommittableHandler<GlobalCommitT> endInputHandler;
+
     /** GlobalCommitT state of this job. Used to filter out previous successful commits. */
     private ListState<GlobalCommitT> streamingCommitterState;
 
+    /** Whether every committer operator completed end input before the restored checkpoint. */
+    private ListState<Boolean> completeEndInputState;
+
     public RestoreCommittableStateManager(
             SerializableSupplier<VersionedSerializer<GlobalCommitT>> committableSerializer,
-            boolean partitionMarkDoneRecoverFromState) {
+            boolean partitionMarkDoneRecoverFromState,
+            EndInputCommittableHandler<GlobalCommitT> endInputHandler) {
         this.committableSerializer = committableSerializer;
         this.partitionMarkDoneRecoverFromState = partitionMarkDoneRecoverFromState;
+        this.endInputHandler = endInputHandler;
     }
 
     @Override
-    public void initializeState(Committer.Context context, Committer<?, GlobalCommitT> committer)
-            throws Exception {
+    public List<GlobalCommitT> initializeState(
+            Committer.Context context, Committer<?, GlobalCommitT> committer) throws Exception {
+        completeEndInputState =
+                context.stateStore()
+                        .getUnionListState(
+                                new ListStateDescriptor<>(
+                                        "streaming_committer_complete_end_input_state",
+                                        BooleanSerializer.INSTANCE));
+        boolean hasCompleteEndInputState = false;
+        boolean restoredCompleteEndInput = true;
+        for (Boolean value : completeEndInputState.get()) {
+            hasCompleteEndInputState = true;
+            restoredCompleteEndInput &= Boolean.TRUE.equals(value);
+        }
+        restoredCompleteEndInput &= hasCompleteEndInputState;
+        completeEndInputState.clear();
+
         streamingCommitterState =
                 new SimpleVersionedListState<>(
                         context.stateStore()
@@ -70,7 +95,26 @@ public void initializeState(Committer.Context context, Committer<?, GlobalCommit
         List<GlobalCommitT> restored = new ArrayList<>();
         streamingCommitterState.get().forEach(restored::add);
         streamingCommitterState.clear();
+
+        List<GlobalCommitT> pendingEndInput = new ArrayList<>();
+        if (!restoredCompleteEndInput) {
+            restored.removeIf(
+                    committable -> {
+                        if (endInputHandler.isEndInput(committable)) {
+                            if (pendingEndInput.isEmpty()) {
+                                pendingEndInput.add(committable);
+                            } else {
+                                pendingEndInput.set(
+                                        0,
+                                        endInputHandler.merge(pendingEndInput.get(0), committable));
+                            }
+                            return true;
+                        }
+                        return false;
+                    });
+        }
         recover(restored, committer);
+        return pendingEndInput;
     }
 
     protected int recover(List<GlobalCommitT> committables, Committer<?, GlobalCommitT> committer)
@@ -79,7 +123,9 @@ protected int recover(List<GlobalCommitT> committables, Committer<?, GlobalCommi
     }
 
     @Override
-    public void snapshotState(List<GlobalCommitT> committables) throws Exception {
+    public void snapshotState(List<GlobalCommitT> committables, boolean completeEndInput)
+            throws Exception {
         streamingCommitterState.update(committables);
+        completeEndInputState.update(Collections.singletonList(completeEndInput));
     }
 }

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

@@ -41,9 +41,32 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /** {@link Committer} for dynamic store. */
 public class StoreCommitter implements Committer<Committable, ManifestCommittable> {
 
+    public static final EndInputCommittableHandler<ManifestCommittable> END_INPUT_HANDLER =
+            new EndInputCommittableHandler<ManifestCommittable>() {
+                private static final long serialVersionUID = 1L;
+
+                @Override
+                public boolean isEndInput(ManifestCommittable committable) {
+                    return committable.identifier() == CommitterOperator.END_INPUT_CHECKPOINT_ID;
+                }
+
+                @Override
+                public ManifestCommittable merge(
+                        ManifestCommittable target, ManifestCommittable source) {
+                    checkArgument(
+                            target.identifier() == source.identifier(),
+                            "Cannot merge committables from different checkpoints %s and %s.",
+                            target.identifier(),
+                            source.identifier());
+                    return mergeManifestCommittables(target, source);
+                }
+            };
+
     private final TableCommitImpl commit;
     @Nullable private final CommitterMetrics committerMetrics;
     private final CommitListeners commitListeners;
@@ -106,6 +129,15 @@ public ManifestCommittable combine(
         return manifestCommittable;
     }
 
+    static ManifestCommittable mergeManifestCommittables(
+            ManifestCommittable target, ManifestCommittable source) {
+        for (CommitMessage commitMessage : source.fileCommittables()) {
+            target.addFileCommittable(commitMessage);
+        }
+        source.properties().forEach(target::addProperty);
+        return target;
+    }
+
     @Override
     public void commit(List<ManifestCommittable> committables)
             throws IOException, InterruptedException {