[GitHub] [flink] dawidwys commented on a change in pull request #14820: [FLINK-21085][runtime][checkpoint] Allows triggering checkpoint for non-source tasks

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

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



##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java
##########
@@ -175,10 +176,23 @@ protected void createInputProcessor(
                         getEnvironment().getTaskInfo());
     }
 
+    protected Optional<CheckpointBarrierHandler> getCheckpointBarrierHandler() {
+        return Optional.ofNullable(checkpointBarrierHandler);
+    }
+
     @Override
     public Future<Boolean> triggerCheckpointAsync(
             CheckpointMetaData metadata, CheckpointOptions options) {
 
+        if (operatorChain.getSourceTaskInputs().size() == 0) {
+            return super.triggerCheckpointAsync(metadata, options);
+        }
+
+        // If there are chained sources, we would always only trigger
+        // the chained sources for checkpoint. This means that for
+        // the checkpoints during the upstream task finished and
+        // this task receives the EndOfPartitionEvent, the checkpoint
+        // would not subsume the pending ones.

Review comment:
       Is that comment correct? We will still subsume the pending checkpoints from within the sources. We do process barriers for the source inputs in that case.

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
##########
@@ -77,6 +77,12 @@
 
     private static final int CHECKPOINT_EXECUTION_DELAY_LOG_THRESHOLD_MS = 30_000;
 
+    /**
+     * TODO Whether enables checkpoints after tasks finished. This is a temporary flag and will be
+     * removed in the last PR.
+     */
+    protected boolean enableCheckpointAfterTasksFinished;

Review comment:
       Why is this `protected`?

##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -1723,6 +1725,86 @@ protected void cancelTask() {
         assertTrue(OpenFailingOperator.wasClosed);
     }
 
+    @Test
+    public void testTriggeringCheckpointWithFinishedChannels() throws Exception {
+        OneInputStreamTaskTestHarness<String, String> testHarness =

Review comment:
       Could you rewrite the test to the newer `StreamTaskMailboxTestHarness`. I believe it should be fairly straightforward.

##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -1723,6 +1725,86 @@ protected void cancelTask() {
         assertTrue(OpenFailingOperator.wasClosed);
     }
 
+    @Test
+    public void testTriggeringCheckpointWithFinishedChannels() throws Exception {
+        OneInputStreamTaskTestHarness<String, String> testHarness =

Review comment:
       Could you rewrite the test to the newer `StreamTaskMailboxTestHarness`. I believe it should be fairly straightforward. Especially as you've used it for the multi-input tasks already.




-- 
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.

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