This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3e1c65ec36a [FLINK-40519][network] Deliver priority barrier to a
credited remote reader while the subpartition is blocked during recovery
3e1c65ec36a is described below
commit 3e1c65ec36aa7725427b39b9affa5709a1f05b5b
Author: Rui Fan <[email protected]>
AuthorDate: Mon Aug 31 17:34:53 2026 +0200
[FLINK-40519][network] Deliver priority barrier to a credited remote reader
while the subpartition is blocked during recovery
---
.../network/partition/PipelinedSubpartition.java | 15 ++++++++--
.../partition/PipelinedSubpartitionTest.java | 34 ++++++++++++++++++++++
2 files changed, 46 insertions(+), 3 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index 2e4b674a3a3..e5369d78506 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -472,7 +472,7 @@ public class PipelinedSubpartition extends
ResultSubpartition implements Channel
// When blocked (e.g. by RECOVERY_COMPLETION event), only allow
priority buffers
// (e.g. unaligned checkpoint barriers) to be polled. Regular
buffers remain blocked
// until resumeConsumption() is called. See
needNotifyPriorityEvent() for details.
- if (isBlocked && buffers.getNumPriorityElements() == 0) {
+ if (isBlockedForDelivery()) {
return null;
}
@@ -619,11 +619,20 @@ public class PipelinedSubpartition extends
ResultSubpartition implements Channel
}
}
+ /**
+ * Blocked for delivery when blocked (e.g. by RECOVERY_COMPLETION) with no
priority element
+ * queued; priority buffers (e.g. unaligned barriers) are still delivered
while blocked.
+ */
+ @GuardedBy("buffers")
+ private boolean isBlockedForDelivery() {
+ return isBlocked && buffers.getNumPriorityElements() == 0;
+ }
+
@GuardedBy("buffers")
private boolean isDataAvailableUnsafe() {
assert Thread.holdsLock(buffers);
- return !isBlocked && (flushRequested || getNumberOfFinishedBuffers() >
0);
+ return !isBlockedForDelivery() && (flushRequested ||
getNumberOfFinishedBuffers() > 0);
}
private Buffer.DataType getNextBufferTypeUnsafe() {
@@ -751,7 +760,7 @@ public class PipelinedSubpartition extends
ResultSubpartition implements Channel
@SuppressWarnings("FieldAccessNotGuarded")
@Override
public int getBuffersInBacklogUnsafe() {
- if (isBlocked || buffers.isEmpty()) {
+ if (isBlockedForDelivery() || buffers.isEmpty()) {
return 0;
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
index 2e8e34d4001..71bc5c200e4 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
@@ -30,6 +30,7 @@ import
org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
import org.apache.flink.runtime.io.network.buffer.BufferConsumer;
+import
org.apache.flink.runtime.io.network.partition.consumer.EndOfOutputChannelStateEvent;
import org.apache.flink.runtime.io.network.util.TestConsumerCallback;
import org.apache.flink.runtime.io.network.util.TestProducerSource;
import org.apache.flink.runtime.io.network.util.TestSubpartitionConsumer;
@@ -472,6 +473,39 @@ public class PipelinedSubpartitionTest extends
SubpartitionTestBase {
.isInstanceOf(ExecutionException.class);
}
+ @TestTemplate
+ void testPriorityBarrierAvailableToCreditedReaderWhileBlocked() throws
Exception {
+ PipelinedSubpartition subpartition = createSubpartition();
+ subpartition.setChannelStateWriter(ChannelStateWriter.NO_OP);
+
+ // Block the subpartition, mirroring the RECOVERY_COMPLETION event
emitted during recovery.
+ subpartition.add(
+
EventSerializer.toBufferConsumer(EndOfOutputChannelStateEvent.INSTANCE, false));
+ pollBufferAndCheckType(subpartition,
Buffer.DataType.RECOVERY_COMPLETION);
+
+ // While blocked and without any priority element, a credited reader
sees no data.
+
assertThat(subpartition.getAvailabilityAndBacklog(true).isAvailable()).isFalse();
+ assertThat(subpartition.pollBuffer()).isNull();
+
+ // Enqueue an unaligned checkpoint barrier as a priority element.
+ CheckpointOptions options =
+ CheckpointOptions.unaligned(
+ CheckpointType.CHECKPOINT,
CheckpointStorageLocationReference.getDefault());
+ subpartition.add(
+ EventSerializer.toBufferConsumer(
+ new CheckpointBarrier(1L, System.currentTimeMillis(),
options), true));
+
+ // The credited-reader availability check must now report available so
the remote reader is
+ // enqueued and the priority barrier is delivered even though the
subpartition
+ // stays blocked.
+
assertThat(subpartition.getAvailabilityAndBacklog(true).isAvailable()).isTrue();
+ pollBufferAndCheckType(subpartition,
Buffer.DataType.PRIORITIZED_EVENT_BUFFER);
+
+ // After the priority element is consumed, the subpartition should
immediately re-block.
+
assertThat(subpartition.getAvailabilityAndBacklog(true).isAvailable()).isFalse();
+ assertThat(subpartition.pollBuffer()).isNull();
+ }
+
private BufferConsumer getTimeoutableBarrierBuffer(long checkpointId)
throws IOException {
CheckpointOptions checkpointOptions =
CheckpointOptions.alignedWithTimeout(
