[
https://issues.apache.org/jira/browse/FLINK-40519?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rui Fan resolved FLINK-40519.
-----------------------------
Fix Version/s: 2.4.0
Resolution: Fixed
Merged to master(2.4.0) via: 3e1c65ec36aa7725427b39b9affa5709a1f05b5b
> 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
> Priority: Major
> Labels: pull-request-available
> Fix For: 2.4.0
>
>
> 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)
