Rui Fan created FLINK-40519:
-------------------------------
Summary: Checkpoint-during-recovery barrier is not delivered to a
credited remote reader while the subpartition is blocked
Key: FLINK-40519
URL: https://issues.apache.org/jira/browse/FLINK-40519
Project: Flink
Issue Type: Sub-task
Components: Runtime / Checkpointing
Reporter: Rui Fan
Assignee: Rui Fan
FLINK-39018
([commit](https://github.com/apache/flink/commit/4f192b3323891adb3b9b8dda7de62d14b901c570))
let a blocked `PipelinedSubpartition` still deliver priority buffers
(unaligned checkpoint barriers), i.e. it is "blocked" only when `isBlocked`
**and** no priority element is queued. But this was applied to `pollBuffer()`
and left out of `isDataAvailableUnsafe()`, which still returns `!isBlocked` [1].
A credited remote reader decides availability through
`isDataAvailableUnsafe()`, so its one-shot priority notification is evaluated
as "not available" and the reader is never enqueued. The barrier is stuck until
`resumeConsumption()` (end of recovery) and the checkpoint can time out.
(No-credit readers and local channels take other paths and are unaffected,
which is why the all-local ITCases miss it.)
Fix: extract the condition once and use it in `pollBuffer()`,
`isDataAvailableUnsafe()` and `getBuffersInBacklogUnsafe()` so they can't drift:
{code:java}
private boolean isBlockedForDelivery() {
return isBlocked && buffers.getNumPriorityElements() == 0;
}
{code}
[1]
https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java#L626
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
