[GitHub] [flink] dawidwys commented on a change in pull request #17055: [FLINK-22367][streaming] Reset syncSavepointId only if it is equal to…

[GitBox <gi...@apache.org>]

dawidwys commented on a change in pull request #17055:
URL: https://github.com/apache/flink/pull/17055#discussion_r700081546



##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -1264,6 +1265,42 @@ public void testCheckpointDeclinedOnClosedOperator() throws Throwable {
         assertEquals(1, harness.getCheckpointResponder().getDeclineReports().size());
     }
 
+    @Test
+    public void testAbortPreviousCheckpointBeforeCompleteTerminateSavepoint() throws Throwable {
+        // given: Marker that the task is finished.
+        AtomicBoolean finishTask = new AtomicBoolean();
+        StreamTaskMailboxTestHarnessBuilder<Integer> builder =
+                new StreamTaskMailboxTestHarnessBuilder<>(
+                                (env) ->
+                                        new OneInputStreamTask<Integer, Integer>(env) {
+                                            @Override
+                                            protected void finishTask() throws Exception {
+                                                super.finishTask();
+                                                finishTask.set(true);
+                                            }
+                                        },
+                                BasicTypeInfo.INT_TYPE_INFO)
+                        .addInput(BasicTypeInfo.INT_TYPE_INFO);
+        StreamTaskMailboxTestHarness<Integer> harness =
+                builder.setupOutputForSingletonOperatorChain(
+                                new TestBoundedOneInputStreamOperator())
+                        .build();
+
+        // when: Receiving the abort notification of the previous checkpoint before the complete
+        // notification of the savepoint terminate.
+        harness.streamTask.notifyCheckpointAbortAsync(1);
+        harness.streamTask.notifyCheckpointCompleteAsync(2);

Review comment:
       How can this happen? You cannot complete a checkpoint that has not been triggered yet. The order of `notifyCheckpointCompleteAsync` and `triggerCheckpointOnBarrier` is wrong here.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org