[release-1.20][FLINK-38370] Ensure CommitterOperator commits all pending committables in batch mode (#27013)

In #26433, we removed the EOI marker in the form of Long.MAX_VALUE as the checkpoint id. Since
streaming pipelines can continue to checkpoint even after their respective operators have been shut
down, it is not safe to use a constant as this can lead to duplicate commits.

However, in batch pipelines we only have one commit on job shutdown. Using any checkpoint id should
suffice in this scenario. Any pending committables should be processed by the ComitterOperator when
the operator shuts down. No further checkpoints will take place.

There are various connectors which rely on this behavior. I don't see any drawbacks from keeping
this behavior for batch pipelines.

Author: mxm

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java

@@ -148,7 +148,7 @@ public void snapshotState(StateSnapshotContext context) throws Exception {
     public void endInput() throws Exception {
         if (!isCheckpointingEnabled || isBatchMode) {
             // There will be no final checkpoint, all committables should be committed here
-            commitAndEmitCheckpoints(lastCompletedCheckpointId + 1);
+            commitAndEmitCheckpoints(Long.MAX_VALUE);
         }
     }
 

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTestBase.java

@@ -276,6 +276,33 @@ void testHandleEndInputInStreamingMode(boolean isCheckpointingEnabled) throws Ex
         assertThat(testHarness.getOutput()).hasSize(2);
     }
 
+    @Test
+    void testEmitCommittablesBatch() throws Exception {
+        SinkAndCounters sinkAndCounters = sinkWithoutPostCommit();
+        final OneInputStreamOperatorTestHarness<
+                        CommittableMessage<String>, CommittableMessage<String>>
+                testHarness =
+                        new OneInputStreamOperatorTestHarness<>(
+                                new CommitterOperatorFactory<>(sinkAndCounters.sink, true, false));
+        testHarness.open();
+
+        // Test that all committables up to Long.MAX_VALUE are committed.
+        long checkpointId = Long.MAX_VALUE;
+        final CommittableSummary<String> committableSummary =
+                new CommittableSummary<>(1, 1, checkpointId, 1, 0, 0);
+        testHarness.processElement(new StreamRecord<>(committableSummary));
+        final CommittableWithLineage<String> committableWithLineage =
+                new CommittableWithLineage<>("1", checkpointId, 1);
+        testHarness.processElement(new StreamRecord<>(committableWithLineage));
+
+        testHarness.endInput();
+
+        assertThat(sinkAndCounters.commitCounter.getAsInt()).isEqualTo(1);
+        assertThat(testHarness.getOutput()).isEmpty();
+
+        testHarness.close();
+    }
+
     private OneInputStreamOperatorTestHarness<
                     CommittableMessage<String>, CommittableMessage<String>>
             createTestHarness(