You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2026/08/13 21:15:18 UTC
(flink) 04/04: [FLINK-40080][network] Remove the unbounded heap fallback from RecoveredInputChannel#requestBufferBlocking
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
commit d9f5e05ea7ad12c029e77610156cdae6b5fda947
Author: Rui Fan <19...@gmail.com>
AuthorDate: Mon Jul 6 02:30:01 2026 +0200
[FLINK-40080][network] Remove the unbounded heap fallback from RecoveredInputChannel#requestBufferBlocking
Back to the plain bufferManager.requestBufferBlocking() form; the OOM
path (and its FLINK-38544 TODO) is gone -- disk spilling supersedes it.
The filtering path no longer competes with itself for network buffers:
the filter output is written to spill files instead of being
re-serialized into freshly requested buffers, so the deadlock the heap
fallback worked around can no longer occur. The handler-side call site
drops the transitional gating parameter accordingly.
---
.../channel/RecoveredChannelStateHandler.java | 4 +--
.../partition/consumer/RecoveredInputChannel.java | 37 ++--------------------
.../consumer/RecoveredInputChannelTest.java | 10 +++---
3 files changed, 7 insertions(+), 44 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
index 23960227843..03ac7f8309c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
@@ -143,9 +143,7 @@ abstract class AbstractInputChannelRecoveredStateHandler
public BufferWithContext<Buffer> getBuffer(InputChannelInfo channelInfo)
throws IOException, InterruptedException {
RecoveredInputChannel channel = getMappedChannels(channelInfo);
- // FLINK-38544 transitional: 'false' bypasses the unbounded heap fallback; the parameter
- // goes away together with the fallback, which disk spilling supersedes.
- Buffer buffer = channel.requestBufferBlocking(false);
+ Buffer buffer = channel.requestBufferBlocking();
return new BufferWithContext<>(wrap(buffer), buffer);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
index 2c6355e8953..4714c4ef28d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java
@@ -19,8 +19,6 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.metrics.Counter;
import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
@@ -29,13 +27,10 @@ import org.apache.flink.runtime.event.TaskEvent;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
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.FreeingBufferRecycler;
-import org.apache.flink.runtime.io.network.buffer.NetworkBuffer;
import org.apache.flink.runtime.io.network.logger.NetworkActionsLogger;
import org.apache.flink.runtime.io.network.partition.ChannelStateHolder;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet;
-import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -289,41 +284,13 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan
}
}
- /**
- * Requests a buffer for reading recovered state. {@code checkpointingDuringRecoveryEnabled} is
- * threaded from the job configuration by the caller now that the gate-level recovery flags are
- * gone; when set, the allocation may fall back to unpooled heap buffers.
- *
- * <p>FLINK-38544 transitional: removed when the spilling backend lands (together with the heap
- * fallback below, which disk spilling supersedes).
- */
- public Buffer requestBufferBlocking(boolean checkpointingDuringRecoveryEnabled)
- throws InterruptedException, IOException {
+ public Buffer requestBufferBlocking() throws InterruptedException, IOException {
// not in setup to avoid assigning buffers unnecessarily if there is no state
if (!exclusiveBuffersAssigned) {
bufferManager.requestExclusiveBuffers(networkBuffersPerChannel);
exclusiveBuffersAssigned = true;
}
- if (!checkpointingDuringRecoveryEnabled) {
- // When checkpoint-during-recovery is not enabled, the original blocking allocation
- // is used as-is — no heap buffer fallback, no behavior change from the legacy path.
- return bufferManager.requestBufferBlocking();
- }
- // Use heap buffer fallback to avoid deadlock during filtering recovery: the filtering
- // thread first requests buffers to read state (pre-filter), then requests more buffers
- // to write filtered output (post-filter). If pre-filter buffers exhaust the pool,
- // post-filter allocation blocks, stalling the thread so pre-filter buffers can never
- // be consumed and released — the thread deadlocks itself. Heap buffers bypass the pool
- // so post-filter writes always proceed. Both call sites (getBuffer and filterAndRewrite)
- // go through this method, so the fallback applies uniformly.
- // TODO: replace heap fallback with disk spilling to bound memory usage in FLINK-38544.
- Buffer buffer = bufferManager.requestBuffer();
- if (buffer != null) {
- return buffer;
- }
- MemorySegment memorySegment =
- MemorySegmentFactory.allocateUnpooledSegment(MemoryManager.DEFAULT_PAGE_SIZE);
- return new NetworkBuffer(memorySegment, FreeingBufferRecycler.INSTANCE);
+ return bufferManager.requestBufferBlocking();
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannelTest.java
index c4fee251a5f..6da822a86ea 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannelTest.java
@@ -166,7 +166,7 @@ class RecoveredInputChannelTest {
RecoveredInputChannel channel = buildPooledChannel(pool, totalSegments);
for (int i = 0; i < totalSegments; i++) {
- channel.requestBufferBlocking(false);
+ channel.requestBufferBlocking();
}
CountDownLatch entered = new CountDownLatch(1);
@@ -176,7 +176,7 @@ class RecoveredInputChannelTest {
() -> {
try {
entered.countDown();
- result.set(channel.requestBufferBlocking(false));
+ result.set(channel.requestBufferBlocking());
} catch (Exception ignored) {
// Thread will be interrupted at teardown.
}
@@ -198,10 +198,8 @@ class RecoveredInputChannelTest {
int totalSegments = 4;
pool = new NetworkBufferPool(totalSegments, MemoryManager.DEFAULT_PAGE_SIZE);
- Buffer filterOnBuf =
- buildPooledChannel(pool, exclusivePerChannel).requestBufferBlocking(true);
- Buffer filterOffBuf =
- buildPooledChannel(pool, exclusivePerChannel).requestBufferBlocking(false);
+ Buffer filterOnBuf = buildPooledChannel(pool, exclusivePerChannel).requestBufferBlocking();
+ Buffer filterOffBuf = buildPooledChannel(pool, exclusivePerChannel).requestBufferBlocking();
// Both must come from the pool — the BufferManager-owned recycler, not the
// FreeingBufferRecycler the heap-fallback used.