[WIP][FLINK-39519] Allocate pre-filter source buffers from heap in filtering mode

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/BufferRequester.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.channel;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/** Supplies per-channel network buffers to the recovery pipeline. */
+@Internal
+interface BufferRequester {
+
+    /** Non-blocking; returns {@code null} when no buffer is currently available. */
+    @Nullable
+    Buffer requestBuffer(InputChannelInfo channelInfo) throws IOException;
+
+    Buffer requestBufferBlocking(InputChannelInfo channelInfo)
+            throws InterruptedException, IOException;
+
+    /**
+     * Releases exclusive buffers for every channel served by this requester. Idempotent. Must run
+     * after the dispatcher's drain has finished so the underlying pools are no longer being read
+     * from.
+     */
+    void releaseExclusiveBuffers() throws IOException;
+}

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java

@@ -26,6 +26,8 @@
 import org.apache.flink.runtime.state.CheckpointStateOutputStream;
 import org.apache.flink.runtime.state.CheckpointStreamFactory;
 import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.IOUtils;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.function.RunnableWithException;
 
@@ -180,6 +182,42 @@ void writeOutput(
         }
     }
 
+    /**
+     * Writes spilled input-channel state chunks as [4-byte length prefix][data bytes], matching the
+     * buffer-based path. The iterator is closed when done.
+     */
+    void writeInputFromSpill(
+            JobVertexID jobVertexID,
+            int subtaskIndex,
+            CloseableIterator<FilteredSpillFile.Chunk> chunks) {
+        if (isDone()) {
+            IOUtils.closeQuietly(chunks);
+            return;
+        }
+        ChannelStatePendingResult pendingResult =
+                getChannelStatePendingResult(jobVertexID, subtaskIndex);
+        runWithChecks(
+                () -> {
+                    checkState(!pendingResult.isAllInputsReceived());
+                    try {
+                        while (chunks.hasNext()) {
+                            FilteredSpillFile.Chunk chunk = chunks.next();
+                            InputChannelInfo info = chunk.getChannelInfo();
+                            long offset = checkpointStream.getPos();
+                            dataStream.writeInt(chunk.getLength());
+                            dataStream.write(chunk.getData(), 0, chunk.getLength());
+                            long size = checkpointStream.getPos() - offset;
+                            pendingResult
+                                    .getInputChannelOffsets()
+                                    .computeIfAbsent(info, unused -> new StateContentMetaInfo())
+                                    .withDataAdded(offset, size);
+                        }
+                    } finally {
+                        chunks.close();
+                    }
+                });
+    }
+
     private <K> void write(
             Map<K, StateContentMetaInfo> offsets,
             K key,