pnowojski commented on a change in pull request #16361:
URL: https://github.com/apache/flink/pull/16361#discussion_r668036271
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointBarrierHandler.java
##########
@@ -153,6 +153,8 @@ protected void markAlignmentEnd() {
}
protected void markAlignmentEnd(long alignmentDuration) {
+ checkState(alignmentDuration >= 0 ||
latestAlignmentDurationNanos.isDone());
Review comment:
> You can take a look at this test
CheckpointBarrierTrackerTest#testCompleteCheckpointsOnLateBarriers, especially
at the last barriers of checkpoint 3 and 4. According to this test, it is
possible to have two last barriers one by one
I think indeed the alignment duration in this case is not calculated
perfectly. For `CheckpointBarrierTracker` ideally we should calculate the
alignment duration independently for each of the checkpoints, but it doesn't
sound like a worthwhile change to me. Anyway, in that case, when we have
checkpoints 3 and 4 closely following one another, `alignmentDuration < 0`
still shouldn't happen. Either for the alignmentDuration we should track only
the oldest or the most recent checkpoint.
As I understand currently we can first call `markAlignmentStart` twice in a
row for chk 3 and chk 4, and after that `markAlignmentEnd` can be called twice
also in a row for both chk 3 and chk 4? And this second `markAlignmentEnd` hits
this condition that the `!latestAlignmentDurationNanos.isDone()` and
`alignmentDuration` can overflow?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
