rkhachatryan commented on a change in pull request #14218:
URL: https://github.com/apache/flink/pull/14218#discussion_r530346023
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -535,13 +536,16 @@ public void checkpointStopped(long checkpointId) {
/**
* Returns a list of buffers, checking the first n non-priority
buffers, and skipping all events.
*/
- private List<Buffer> getInflightBuffersUnsafe(long checkpointId) {
+ private List<Buffer> getInflightBuffersUnsafe(long checkpointId) throws
CheckpointException {
assert Thread.holdsLock(receivedBuffers);
+ if (checkpointId < lastBarrierId) {
+ throw new CheckpointException(
Review comment:
I was in doubt whether this exception type is appropriate here or
whether an `Optional` should be returned.
But at the upper level, it ultimately should be used, and the reason is only
known here (so it would be harder to analyze the exception/empty buffers later).
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/SingleCheckpointBarrierHandler.java
##########
@@ -158,14 +163,16 @@ public void processBarrierAnnouncement(
@Override
public void processCancellationBarrier(CancelCheckpointMarker
cancelBarrier) throws IOException {
final long cancelledId = cancelBarrier.getCheckpointId();
- if (currentCheckpointId > cancelledId || (currentCheckpointId
== cancelledId && numBarriersReceived == 0)) {
- return;
+ if (cancelledId >= currentCheckpointId && (cancelledId >
currentCheckpointId || numBarriersReceived > 0)) {
+ abortInternal(cancelledId, new
CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED_ON_CANCELLATION_BARRIER));
}
+ }
+
+ private void abortInternal(long cancelledId, CheckpointException
exception) throws IOException {
// by setting the currentCheckpointId to this checkpoint while
keeping the numBarriers
// at zero means that no checkpoint barrier can start a new
alignment
- currentCheckpointId = cancelledId;
+ currentCheckpointId = Math.max(cancelledId,
currentCheckpointId);
Review comment:
This is not necessary currently.
But because I extracted this method the assignment became more error-prone,
so I added this check.
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/SingleCheckpointBarrierHandler.java
##########
@@ -158,14 +163,16 @@ public void processBarrierAnnouncement(
@Override
public void processCancellationBarrier(CancelCheckpointMarker
cancelBarrier) throws IOException {
final long cancelledId = cancelBarrier.getCheckpointId();
- if (currentCheckpointId > cancelledId || (currentCheckpointId
== cancelledId && numBarriersReceived == 0)) {
- return;
+ if (cancelledId >= currentCheckpointId && (cancelledId >
currentCheckpointId || numBarriersReceived > 0)) {
Review comment:
Please double-check this @AHeise :)
----------------------------------------------------------------
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:
[email protected]
