[FLINK-37747][runtime] Use old subtask count for restored committable objects.

* [FLINK-37747][runtime] Use old subtask count for restored committable objects

* trying to adapt test

* Attempt to backport SinkV2ITCase from FLINK-37747

* Add more test use cases for SinkV2ITCase

* Fix SinkV2ITCase test

* Setup TestSinkV2WithPostCommitTopology global committer

* Rename test SinkV2ITCase streaming source to be more specific

---------

Co-authored-by: Eric Nascimento <[email protected]>
Co-authored-by: David Wang <[email protected]>

Author: 3 people

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

@@ -180,7 +180,12 @@ private void commitAndEmit(CheckpointCommittableManager<CommT> committableManage
 
     private void emit(CheckpointCommittableManager<CommT> committableManager) {
         int subtaskId = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
-        int numberOfSubtasks = getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks();
+        // Ensure that numberOfSubtasks is in sync with the number of actually emitted
+        // CommittableSummaries during upscaling recovery (see FLINK-37747).
+        int numberOfSubtasks =
+                Math.min(
+                        getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks(),
+                        committableManager.getNumberOfSubtasks());
         long checkpointId = committableManager.getCheckpointId();
         Collection<CommT> committables = committableManager.getSuccessfulCommittables();
         output.collect(

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

@@ -31,6 +31,7 @@
 import org.assertj.core.api.ListAssert;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.util.function.IntSupplier;
@@ -91,7 +92,7 @@ void ensureAllCommittablesArrivedBeforeCommitting() throws Exception {
         SinkAndCounters sinkAndCounters = sinkWithPostCommit();
         final OneInputStreamOperatorTestHarness<
                         CommittableMessage<String>, CommittableMessage<String>>
-                testHarness = createTestHarness(sinkAndCounters.sink, false, true);
+                testHarness = createTestHarness(sinkAndCounters.sink, false, true, 1, 1, 0);
         testHarness.open();
         testHarness.setProcessingTime(0);
 
@@ -125,11 +126,13 @@ void ensureAllCommittablesArrivedBeforeCommitting() throws Exception {
         testHarness.close();
     }
 
-    @Test
-    void testStateRestore() throws Exception {
+    @ParameterizedTest
+    @CsvSource({"1, 10, 9", "2, 1, 0", "2, 2, 1"})
+    void testStateRestoreWithScaling(
+            int parallelismBeforeScaling, int parallelismAfterScaling, int subtaskIdAfterRecovery)
+            throws Exception {
 
         final int originalSubtaskId = 0;
-        final int subtaskIdAfterRecovery = 9;
 
         final OneInputStreamOperatorTestHarness<
                         CommittableMessage<String>, CommittableMessage<String>>
@@ -138,8 +141,8 @@ void testStateRestore() throws Exception {
                                 sinkWithPostCommitWithRetry().sink,
                                 false,
                                 true,
-                                1,
-                                1,
+                                parallelismBeforeScaling,
+                                parallelismBeforeScaling,
                                 originalSubtaskId);
         testHarness.open();
 
@@ -148,15 +151,17 @@ void testStateRestore() throws Exception {
         long checkpointId = 0L;
 
         final CommittableSummary<String> committableSummary =
-                new CommittableSummary<>(originalSubtaskId, 1, checkpointId, 1, 1, 0);
+                new CommittableSummary<>(
+                        originalSubtaskId, parallelismBeforeScaling, checkpointId, 1, 1, 0);
         testHarness.processElement(new StreamRecord<>(committableSummary));
         final CommittableWithLineage<String> first =
                 new CommittableWithLineage<>("1", checkpointId, originalSubtaskId);
         testHarness.processElement(new StreamRecord<>(first));
 
         // another committable for the same checkpointId but from different subtask.
         final CommittableSummary<String> committableSummary2 =
-                new CommittableSummary<>(originalSubtaskId + 1, 1, checkpointId, 1, 1, 0);
+                new CommittableSummary<>(
+                        originalSubtaskId + 1, parallelismBeforeScaling, checkpointId, 1, 1, 0);
         testHarness.processElement(new StreamRecord<>(committableSummary2));
         final CommittableWithLineage<String> second =
                 new CommittableWithLineage<>("2", checkpointId, originalSubtaskId + 1);
@@ -174,7 +179,12 @@ void testStateRestore() throws Exception {
                         CommittableMessage<String>, CommittableMessage<String>>
                 restored =
                         createTestHarness(
-                                sinkAndCounters.sink, false, true, 10, 10, subtaskIdAfterRecovery);
+                                sinkAndCounters.sink,
+                                false,
+                                true,
+                                parallelismAfterScaling,
+                                parallelismAfterScaling,
+                                subtaskIdAfterRecovery);
 
         restored.initializeState(snapshot);
         restored.open();
@@ -188,7 +198,8 @@ void testStateRestore() throws Exception {
                 .hasSubtaskId(subtaskIdAfterRecovery)
                 .hasFailedCommittables(0)
                 .hasOverallCommittables(2)
-                .hasPendingCommittables(0);
+                .hasPendingCommittables(0)
+                .hasNumberOfSubtasks(Math.min(parallelismBeforeScaling, parallelismAfterScaling));
 
         // Expect the same checkpointId that the original snapshot was made with.
         records.element(1, as(committableWithLineage()))
@@ -303,17 +314,6 @@ void testEmitCommittablesBatch() throws Exception {
         testHarness.close();
     }
 
-    private OneInputStreamOperatorTestHarness<
-                    CommittableMessage<String>, CommittableMessage<String>>
-            createTestHarness(
-                    SupportsCommitter<String> sink,
-                    boolean isBatchMode,
-                    boolean isCheckpointingEnabled)
-                    throws Exception {
-        return new OneInputStreamOperatorTestHarness<>(
-                new CommitterOperatorFactory<>(sink, isBatchMode, isCheckpointingEnabled));
-    }
-
     private OneInputStreamOperatorTestHarness<
                     CommittableMessage<String>, CommittableMessage<String>>
             createTestHarness(

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

@@ -243,7 +243,7 @@ public TestSinkV2WithPostCommitTopology(
         @Override
         public void addPostCommitTopology(DataStream<CommittableMessage<String>> committables) {
             StandardSinkTopologies.addGlobalCommitter(
-                    committables, DefaultCommitter::new, this::getCommittableSerializer);
+                    committables, () -> createCommitter(null), this::getCommittableSerializer);
         }
     }