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 <[email protected]> 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.
