[GitHub] [flink] pnowojski commented on a change in pull request #14797: [FLINK-21104][network] Various fixes for UC

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

pnowojski commented on a change in pull request #14797:
URL: https://github.com/apache/flink/pull/14797#discussion_r566825724



##########
File path: flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java
##########
@@ -203,6 +203,7 @@ public Boundedness getBoundedness() {
             private int numAbortedCheckpoints;
             private boolean throttle = true;
             private int numRestarts;
+            private boolean shouldFinish = false;

Review comment:
       maybe replace `shouldFinish` and `throttle` with an enum state?

##########
File path: flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java
##########
@@ -282,20 +281,39 @@ public void addSplits(List<LongSplit> splits) {
                             "Tried to add " + splits + " but already got " + split);
                 }
                 split = Iterables.getOnlyElement(splits);
+                checkShouldFinish();
                 LOG.info(
-                        "Added split {} @ {} subtask ({} attempt)",
+                        "Added split {}, shouldFinish={} @ {} subtask ({} attempt)",
                         split,
+                        shouldFinish,
                         split.nextNumber % split.increment,
                         numRestarts);
             }
 
+            /** Should only be called if the split has been successfully checkpointed. */
+            private void checkShouldFinish() {
+                shouldFinish =
+                        split != null
+                                && split.numCompletedCheckpoints >= minCheckpoints
+                                && numRestarts >= expectedRestarts;
+            }
+
             @Override
-            public void notifyNoMoreSplits() {}
+            public void notifyNoMoreSplits() {
+                LOG.info("notifyNoMoreSplits ({} attempt)", numRestarts);
+                shouldFinish = true;
+            }

Review comment:
       Is this being used?

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java
##########
@@ -109,7 +109,9 @@ private void processPriorityEvents() throws IOException, InterruptedException {
         while (hasPriorityEvent) {
             // process as many priority events as possible
             final Optional<BufferOrEvent> bufferOrEventOpt = pollNext();
-            checkState(bufferOrEventOpt.isPresent());
+            if (!bufferOrEventOpt.isPresent()) {
+                break;
+            }

Review comment:
       I think this is missing a test. In `CheckpointedInputGateTest`?




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