[GitHub] [flink] AHeise commented on a change in pull request #14807: [FLINK-20654][FLINK-21104][network] Fix two bugs in the handling of UnalignedCheckpoints

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

AHeise commented on a change in pull request #14807:
URL: https://github.com/apache/flink/pull/14807#discussion_r566953027



##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/ChannelStatePersisterTest.java
##########
@@ -65,38 +83,58 @@ public void testNewBarrierNotOverwrittenByCheckForBarrier() throws IOException {
     }
 
     @Test
-    public void testLateBarrierOnCancelledCheckpoint() throws IOException {
-        ChannelStatePersister persister =
-                new ChannelStatePersister(ChannelStateWriter.NO_OP, new InputChannelInfo(0, 0));
+    public void testLateBarrierOnStartedAndCancelledCheckpoint() throws Exception {
+        testLateBarrier(true, true);
+    }
 
-        persister.startPersisting(1L, Collections.emptyList());
-        // checkpoint aborted
-        persister.stopPersisting(1L);
+    @Test
+    public void testLateBarrierOnCancelledCheckpoint() throws Exception {
+        testLateBarrier(false, true);
+    }
 
-        // late barrier
-        persister.checkForBarrier(barrier(1L));
+    @Test
+    public void testLateBarrierOnNotYetCancelledCheckpoint() throws Exception {
+        testLateBarrier(false, false);
+    }
 
-        persister.startPersisting(2L, Collections.emptyList());
-        persister.checkForBarrier(barrier(2L));
+    public void testLateBarrier(

Review comment:
       nit: private

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -638,6 +637,11 @@ private void notifyPriorityEventForce() {
                     CheckpointFailureReason
                             .CHECKPOINT_SUBSUMED); // currently, at most one active unaligned
             // checkpoint is possible
+        } else if (checkpointId > lastBarrierId) {
+            // This channel has received some obsolete barrier, older compared to the checkpointId
+            // which we are processing right now, and we should ignore that obsoleted checkpoint
+            // barrier sequence number.
+            resetLastBarrier();

Review comment:
       I wouldn't have had a side-effect here and just returned an empty list (which I'd also return with side-effect).

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/ChannelStatePersister.java
##########
@@ -67,16 +68,27 @@
         this.channelInfo = checkNotNull(channelInfo);
     }
 
-    protected void startPersisting(long barrierId, List<Buffer> knownBuffers) {
+    protected void startPersisting(long barrierId, List<Buffer> knownBuffers)
+            throws CheckpointException {
         logEvent("startPersisting", barrierId);
-        if (checkpointStatus != CheckpointStatus.BARRIER_RECEIVED && lastSeenBarrier < barrierId) {
+        if (checkpointStatus == CheckpointStatus.BARRIER_RECEIVED && lastSeenBarrier > barrierId) {
+            throw new CheckpointException(

Review comment:
       How does it guarantee that this doesn't cancel a newer checkpoint? I don't see any checkpoint id being bound to the exception.




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