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 2e4b674a3a3abd..e5369d78506d11 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 @@ BufferAndBacklog pollBuffer() {
             // 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 ResultSubpartitionView.AvailabilityWithBacklog getAvailabilityAndBacklog(
         }
     }
 
+    /**
+     * 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 @@ private void increaseBuffersInBacklog(BufferConsumer buffer) {
     @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 2e8e34d4001464..8c5624e4bedbd3 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.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,35 @@ void testConcurrentTimeoutableCheckpointBarrier() throws Exception {
                 .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);
+    }
+
     private BufferConsumer getTimeoutableBarrierBuffer(long checkpointId) throws IOException {
         CheckpointOptions checkpointOptions =
                 CheckpointOptions.alignedWithTimeout(