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/07/17 14:19:28 UTC
(flink) 02/03: [FLINK-39522][network] Remove recovery flags and the filtering-complete future from the gate API
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 2f21971b7d5bb6963270968b6b17ee084ca7089a
Author: Rui Fan <19...@gmail.com>
AuthorDate: Mon Jul 6 00:59:46 2026 +0200
[FLINK-39522][network] Remove recovery flags and the filtering-complete future from the gate API
Pure deletion of the now-unused v1 gating: the previous commit made StreamTask
gate the flag-on RUNNING transition on the recovery future chain instead of the
filtering-progress future, so the API has no callers left.
Drop InputGate#getBufferFilteringCompleteFuture and its overrides in
SingleInputGate, UnionInputGate and InputGateWithMetrics;
IndexedInputGate#setCheckpointingDuringRecoveryEnabled /
#isCheckpointingDuringRecoveryEnabled together with SingleInputGate's volatile
backing field; and RecoveredInputChannel's bufferFilteringCompleteFuture field,
getter and the finishReadRecoveredState lock-ordering commentary tied to it.
The one remaining consumer of the flag -- the unbounded heap-buffer fallback in
RecoveredInputChannel#requestBufferBlocking, which stays until disk spilling
supersedes it -- now has the flag threaded explicitly from the job
configuration through the recovery-handler call sites
(requestBufferBlocking(boolean)) instead of reading it off the gate.
Remove the corresponding testBufferFilteringCompleteFutureAggregation cases in
SingleInputGateTest/UnionInputGateTest and adapt MockInputGate,
MockIndexedInputGate, SingleInputGateBuilder and
AlignedCheckpointsMassiveRandomTest.
Co-authored-by: Roman Khachatryan <kh...@gmail.com>
---
.../channel/RecoveredChannelStateHandler.java | 29 +++++++----
.../partition/consumer/IndexedInputGate.java | 6 ---
.../io/network/partition/consumer/InputGate.java | 7 ---
.../partition/consumer/RecoveredInputChannel.java | 42 +++++-----------
.../partition/consumer/SingleInputGate.java | 27 -----------
.../network/partition/consumer/UnionInputGate.java | 9 ----
.../runtime/taskmanager/InputGateWithMetrics.java | 15 ------
.../partition/consumer/SingleInputGateBuilder.java | 8 ----
.../partition/consumer/SingleInputGateTest.java | 30 ------------
.../partition/consumer/UnionInputGateTest.java | 56 ----------------------
.../streaming/runtime/io/MockIndexedInputGate.java | 13 -----
.../flink/streaming/runtime/io/MockInputGate.java | 13 -----
.../AlignedCheckpointsMassiveRandomTest.java | 13 -----
13 files changed, 32 insertions(+), 236 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 64b39d1f5f1..72d0b995f29 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
@@ -90,10 +90,20 @@ abstract class AbstractInputChannelRecoveredStateHandler
final Map<InputChannelInfo, RecoveredInputChannel> rescaledChannels = new HashMap<>();
final Map<Integer, RescaleMappings> oldToNewMappings = new HashMap<>();
+ /**
+ * FLINK-38544 transitional: removed when the spilling backend lands. Gates the unbounded
+ * heap-buffer fallback in {@link RecoveredInputChannel#requestBufferBlocking(boolean)}; read
+ * from the job configuration at the call site now that the gate-level recovery flags are gone.
+ */
+ final boolean checkpointingDuringRecoveryEnabled;
+
AbstractInputChannelRecoveredStateHandler(
- InputGate[] inputGates, InflightDataRescalingDescriptor channelMapping) {
+ InputGate[] inputGates,
+ InflightDataRescalingDescriptor channelMapping,
+ boolean checkpointingDuringRecoveryEnabled) {
this.inputGates = inputGates;
this.channelMapping = channelMapping;
+ this.checkpointingDuringRecoveryEnabled = checkpointingDuringRecoveryEnabled;
}
/**
@@ -113,12 +123,12 @@ abstract class AbstractInputChannelRecoveredStateHandler
@Nullable ChannelStateFilteringHandler filteringHandler,
int memorySegmentSize) {
if (!checkpointingDuringRecoveryEnabled) {
- return new NoSpillingHandler(inputGates, channelMapping);
+ return new NoSpillingHandler(inputGates, channelMapping, false);
}
// FLINK-38544 transitional: the flag-on path still uses the in-memory handlers until the
// spilling backend lands.
if (filteringHandler == null) {
- return new NoSpillingHandler(inputGates, channelMapping);
+ return new NoSpillingHandler(inputGates, channelMapping, true);
}
return new FilteringHandler(
inputGates, channelMapping, filteringHandler, memorySegmentSize);
@@ -129,7 +139,7 @@ abstract class AbstractInputChannelRecoveredStateHandler
public BufferWithContext<Buffer> getBuffer(InputChannelInfo channelInfo)
throws IOException, InterruptedException {
RecoveredInputChannel channel = getMappedChannels(channelInfo);
- Buffer buffer = channel.requestBufferBlocking();
+ Buffer buffer = channel.requestBufferBlocking(checkpointingDuringRecoveryEnabled);
return new BufferWithContext<>(wrap(buffer), buffer);
}
@@ -174,8 +184,11 @@ abstract class AbstractInputChannelRecoveredStateHandler
*/
class NoSpillingHandler extends AbstractInputChannelRecoveredStateHandler {
- NoSpillingHandler(InputGate[] inputGates, InflightDataRescalingDescriptor channelMapping) {
- super(inputGates, channelMapping);
+ NoSpillingHandler(
+ InputGate[] inputGates,
+ InflightDataRescalingDescriptor channelMapping,
+ boolean checkpointingDuringRecoveryEnabled) {
+ super(inputGates, channelMapping, checkpointingDuringRecoveryEnabled);
}
@Override
@@ -237,7 +250,7 @@ class FilteringHandler extends AbstractInputChannelRecoveredStateHandler {
InflightDataRescalingDescriptor channelMapping,
ChannelStateFilteringHandler filteringHandler,
int memorySegmentSize) {
- super(inputGates, channelMapping);
+ super(inputGates, channelMapping, true);
this.filteringHandler = filteringHandler;
checkArgument(
memorySegmentSize > 0, "memorySegmentSize must be positive: %s", memorySegmentSize);
@@ -307,7 +320,7 @@ class FilteringHandler extends AbstractInputChannelRecoveredStateHandler {
oldSubtaskIndex,
channelInfo.getInputChannelIdx(),
retainedBuffer,
- channel::requestBufferBlocking);
+ () -> channel.requestBufferBlocking(checkpointingDuringRecoveryEnabled));
int i = 0;
try {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/IndexedInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/IndexedInputGate.java
index 915012924d1..5daa277cd9b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/IndexedInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/IndexedInputGate.java
@@ -74,10 +74,4 @@ public abstract class IndexedInputGate extends InputGate implements Checkpointab
public abstract ResultPartitionType getConsumedPartitionType();
public abstract void triggerDebloating();
-
- /** Sets whether unaligned checkpointing during recovery is enabled. */
- public abstract void setCheckpointingDuringRecoveryEnabled(boolean enabled);
-
- /** Returns whether unaligned checkpointing during recovery is enabled. */
- public abstract boolean isCheckpointingDuringRecoveryEnabled();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java
index 2348c33ae98..142f45cf9bc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java
@@ -209,12 +209,5 @@ public abstract class InputGate
public abstract CompletableFuture<Void> getStateConsumedFuture();
- /**
- * Returns a future that completes when buffer filtering is complete for all channels. This
- * future completes before {@link #getStateConsumedFuture()}, enabling earlier RUNNING state
- * transition when unaligned checkpoint during recovery is enabled.
- */
- public abstract CompletableFuture<Void> getBufferFilteringCompleteFuture();
-
public abstract void finishReadRecoveredState() throws IOException;
}
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 ec878b69c62..04a5b65c2f1 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
@@ -62,13 +62,6 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan
private final CompletableFuture<Void> stateConsumedFuture = new CompletableFuture<>();
protected final BufferManager bufferManager;
- /**
- * Future that completes when recovered buffers have been filtered for this channel. This
- * completes before stateConsumedFuture, enabling earlier RUNNING state transition when
- * unaligned checkpoint during recovery is enabled.
- */
- private final CompletableFuture<Void> bufferFilteringCompleteFuture = new CompletableFuture<>();
-
@GuardedBy("receivedBuffers")
private boolean isReleased;
@@ -178,14 +171,6 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan
protected abstract InputChannel toInputChannelInternal(boolean needsRecovery)
throws IOException;
- /**
- * Returns the future that completes when buffer filtering is complete. This future completes
- * before stateConsumedFuture, at the point when finishReadRecoveredState() is called.
- */
- CompletableFuture<Void> getBufferFilteringCompleteFuture() {
- return bufferFilteringCompleteFuture;
- }
-
@Override
public CompletableFuture<Void> getStateConsumedFuture() {
return stateConsumedFuture;
@@ -220,21 +205,9 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan
}
public void finishReadRecoveredState() throws IOException {
- // Adding the event and completing the future must be atomic under receivedBuffers lock.
- // Without this, either ordering has a race:
- // - event first: task thread consumes EndOfInputChannelStateEvent, which completes
- // stateConsumedFuture. When checkpointing during recovery is disabled,
- // stateConsumedFuture triggers requestPartitions -> toInputChannel(), which
- // fails because bufferFilteringCompleteFuture is not yet done.
- // - future first: toInputChannel() extracts buffers before the event is added,
- // losing the EndOfInputChannelStateEvent.
- // Both toInputChannel() and getNextRecoveredStateBuffer() synchronize on
- // receivedBuffers, so holding the same lock here guarantees
- // bufferFilteringCompleteFuture is always done before stateConsumedFuture.
synchronized (receivedBuffers) {
onRecoveredStateBuffer(
EventSerializer.toBuffer(EndOfInputChannelStateEvent.INSTANCE, false));
- bufferFilteringCompleteFuture.complete(null);
}
bufferManager.releaseFloatingBuffers();
LOG.debug("{}/{} finished recovering input.", inputGate.getOwningTaskName(), channelInfo);
@@ -254,8 +227,6 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan
if (next == null) {
return null;
} else if (isEndOfInputChannelStateEvent(next)) {
- Preconditions.checkState(
- bufferFilteringCompleteFuture.isDone(), "buffer filtering is not complete");
stateConsumedFuture.complete(null);
return null;
} else {
@@ -357,13 +328,22 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan
}
}
- public Buffer requestBufferBlocking() throws InterruptedException, IOException {
+ /**
+ * 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 {
// not in setup to avoid assigning buffers unnecessarily if there is no state
if (!exclusiveBuffersAssigned) {
bufferManager.requestExclusiveBuffers(networkBuffersPerChannel);
exclusiveBuffersAssigned = true;
}
- if (!inputGate.isCheckpointingDuringRecoveryEnabled()) {
+ 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();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
index 4a7c9c6617e..58856eb4761 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
@@ -243,8 +243,6 @@ public class SingleInputGate extends IndexedInputGate {
*/
private final int[] endOfPartitions;
- private volatile boolean checkpointingDuringRecoveryEnabled = false;
-
public SingleInputGate(
String owningTaskName,
int gateIndex,
@@ -329,31 +327,6 @@ public class SingleInputGate extends IndexedInputGate {
}
}
- @Override
- public void setCheckpointingDuringRecoveryEnabled(boolean enabled) {
- this.checkpointingDuringRecoveryEnabled = enabled;
- }
-
- @Override
- public boolean isCheckpointingDuringRecoveryEnabled() {
- return checkpointingDuringRecoveryEnabled;
- }
-
- @Override
- public CompletableFuture<Void> getBufferFilteringCompleteFuture() {
- synchronized (requestLock) {
- List<CompletableFuture<?>> futures = new ArrayList<>(numberOfInputChannels);
- for (InputChannel inputChannel : inputChannels()) {
- if (inputChannel instanceof RecoveredInputChannel) {
- futures.add(
- ((RecoveredInputChannel) inputChannel)
- .getBufferFilteringCompleteFuture());
- }
- }
- return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
- }
- }
-
@Override
public void requestPartitions() {
requestPartitions(false);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
index 37aee532ae5..b7a708f38f2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGate.java
@@ -357,15 +357,6 @@ public class UnionInputGate extends InputGate {
.toArray(new CompletableFuture[] {}));
}
- @Override
- public CompletableFuture<Void> getBufferFilteringCompleteFuture() {
- return CompletableFuture.allOf(
- inputGatesByGateIndex.values().stream()
- .map(InputGate::getBufferFilteringCompleteFuture)
- .collect(Collectors.toList())
- .toArray(new CompletableFuture[] {}));
- }
-
@Override
public void requestPartitions() throws IOException {
requestPartitions(false);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java
index 695d897c92f..5d1ca9c680b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/InputGateWithMetrics.java
@@ -125,11 +125,6 @@ public class InputGateWithMetrics extends IndexedInputGate {
return inputGate.getStateConsumedFuture();
}
- @Override
- public CompletableFuture<Void> getBufferFilteringCompleteFuture() {
- return inputGate.getBufferFilteringCompleteFuture();
- }
-
@Override
public void requestPartitions() throws IOException {
inputGate.requestPartitions();
@@ -175,16 +170,6 @@ public class InputGateWithMetrics extends IndexedInputGate {
inputGate.finishReadRecoveredState();
}
- @Override
- public void setCheckpointingDuringRecoveryEnabled(boolean enabled) {
- inputGate.setCheckpointingDuringRecoveryEnabled(enabled);
- }
-
- @Override
- public boolean isCheckpointingDuringRecoveryEnabled() {
- return inputGate.isCheckpointingDuringRecoveryEnabled();
- }
-
private BufferOrEvent updateMetrics(BufferOrEvent bufferOrEvent) {
int incomingDataSize = bufferOrEvent.getSize();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
index a4da811f8a3..e4a4c289dc6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateBuilder.java
@@ -83,8 +83,6 @@ public class SingleInputGateBuilder {
private TieredStorageConsumerClient tieredStorageConsumerClient = null;
- private boolean isCheckpointingDuringRecoveryEnabled = false;
-
public SingleInputGateBuilder setPartitionProducerStateProvider(
PartitionProducerStateProvider partitionProducerStateProvider) {
@@ -169,11 +167,6 @@ public class SingleInputGateBuilder {
return this;
}
- public SingleInputGateBuilder setCheckpointingDuringRecoveryEnabled(boolean enabled) {
- this.isCheckpointingDuringRecoveryEnabled = enabled;
- return this;
- }
-
public SingleInputGate build() {
SingleInputGate gate =
new SingleInputGate(
@@ -202,7 +195,6 @@ public class SingleInputGateBuilder {
.toArray(InputChannel[]::new));
}
gate.setTieredStorageService(null, tieredStorageConsumerClient, null);
- gate.setCheckpointingDuringRecoveryEnabled(isCheckpointingDuringRecoveryEnabled);
return gate;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
index b2cc9d7ce3c..f7f0b744fb9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGateTest.java
@@ -142,36 +142,6 @@ class SingleInputGateTest extends InputGateTestBase {
.isInstanceOf(CheckpointException.class);
}
- @Test
- void testBufferFilteringCompleteFutureAggregation() throws Exception {
- final NettyShuffleEnvironment environment = createNettyShuffleEnvironment();
- final SingleInputGate inputGate = createInputGate(environment);
- try (Closer closer = Closer.create()) {
- closer.register(environment::close);
- closer.register(inputGate::close);
-
- // Enable unaligned during recovery for this test so that
- // bufferFilteringCompleteFuture is completed by finishReadRecoveredState()
- inputGate.setCheckpointingDuringRecoveryEnabled(true);
- inputGate.setup();
-
- // Initially, the aggregated future should not be completed
- assertThat(inputGate.getBufferFilteringCompleteFuture()).isNotDone();
-
- // After finishing read recovered state, bufferFilteringCompleteFuture should be
- // completed (only when config is enabled)
- inputGate.finishReadRecoveredState();
- assertThat(inputGate.getBufferFilteringCompleteFuture()).isDone();
-
- // stateConsumedFuture should not be completed until data is consumed
- assertThat(inputGate.getStateConsumedFuture()).isNotDone();
-
- // Consuming the EndOfInputChannelStateEvent should complete stateConsumedFuture
- inputGate.pollNext();
- assertThat(inputGate.getStateConsumedFuture()).isDone();
- }
- }
-
/**
* Tests {@link InputGate#setup()} should create the respective {@link BufferPool} and assign
* exclusive buffers for {@link RemoteInputChannel}s, but should not request partitions.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
index a7325d15375..419246137e8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/UnionInputGateTest.java
@@ -18,14 +18,12 @@
package org.apache.flink.runtime.io.network.partition.consumer;
-import org.apache.flink.metrics.SimpleCounter;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.io.PullingAsyncDataInput;
import org.apache.flink.runtime.io.network.api.StopMode;
import org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
import org.apache.flink.runtime.io.network.partition.NoOpResultSubpartitionView;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
-import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet;
import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateTest.TestingResultPartitionManager;
import org.junit.jupiter.api.Test;
@@ -277,60 +275,6 @@ class UnionInputGateTest extends InputGateTestBase {
assertThat(unionInputGate.getChannel(1)).isEqualTo(inputChannel2);
}
- @Test
- void testBufferFilteringCompleteFutureAggregation() throws IOException {
- // Create 2 SingleInputGates, each with 1 RecoveredInputChannel
- SingleInputGate ig1 =
- new SingleInputGateBuilder().setCheckpointingDuringRecoveryEnabled(true).build();
- RecoveredInputChannel channel1 = buildRecoveredChannel(ig1);
- ig1.setInputChannels(channel1);
-
- SingleInputGate ig2 =
- new SingleInputGateBuilder()
- .setSingleInputGateIndex(1)
- .setCheckpointingDuringRecoveryEnabled(true)
- .build();
- RecoveredInputChannel channel2 = buildRecoveredChannel(ig2);
- ig2.setInputChannels(channel2);
-
- UnionInputGate union = new UnionInputGate(ig1, ig2);
-
- // Initially, bufferFilteringCompleteFuture should not be done
- assertThat(union.getBufferFilteringCompleteFuture()).isNotDone();
- assertThat(union.getStateConsumedFuture()).isNotDone();
-
- // Complete buffer filtering on first gate only
- channel1.finishReadRecoveredState();
- assertThat(ig1.getBufferFilteringCompleteFuture()).isDone();
- assertThat(union.getBufferFilteringCompleteFuture()).isNotDone();
-
- // Complete buffer filtering on second gate
- channel2.finishReadRecoveredState();
- assertThat(ig2.getBufferFilteringCompleteFuture()).isDone();
- assertThat(union.getBufferFilteringCompleteFuture()).isDone();
-
- // State consumed futures should still NOT be done (state not consumed yet)
- assertThat(union.getStateConsumedFuture()).isNotDone();
- }
-
- private static RecoveredInputChannel buildRecoveredChannel(SingleInputGate inputGate) {
- return new RecoveredInputChannel(
- inputGate,
- 0,
- new ResultPartitionID(),
- new ResultSubpartitionIndexSet(0),
- 0,
- 0,
- new SimpleCounter(),
- new SimpleCounter(),
- 10) {
- @Override
- protected InputChannel toInputChannelInternal(boolean needsRecovery) {
- throw new UnsupportedOperationException();
- }
- };
- }
-
@Test
void testEmptyPull() throws IOException, InterruptedException {
final SingleInputGate inputGate1 = createInputGate(1);
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockIndexedInputGate.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockIndexedInputGate.java
index 5cc44377707..53bba67f7a2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockIndexedInputGate.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockIndexedInputGate.java
@@ -57,11 +57,6 @@ public class MockIndexedInputGate extends IndexedInputGate {
return CompletableFuture.completedFuture(null);
}
- @Override
- public CompletableFuture<Void> getBufferFilteringCompleteFuture() {
- return CompletableFuture.completedFuture(null);
- }
-
@Override
public void finishReadRecoveredState() {}
@@ -147,12 +142,4 @@ public class MockIndexedInputGate extends IndexedInputGate {
@Override
public void triggerDebloating() {}
-
- @Override
- public void setCheckpointingDuringRecoveryEnabled(boolean enabled) {}
-
- @Override
- public boolean isCheckpointingDuringRecoveryEnabled() {
- return false;
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockInputGate.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockInputGate.java
index e63a49be53b..47e3b79a77f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockInputGate.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/MockInputGate.java
@@ -80,11 +80,6 @@ public class MockInputGate extends IndexedInputGate {
return CompletableFuture.completedFuture(null);
}
- @Override
- public CompletableFuture<Void> getBufferFilteringCompleteFuture() {
- return CompletableFuture.completedFuture(null);
- }
-
@Override
public void finishReadRecoveredState() {}
@@ -209,12 +204,4 @@ public class MockInputGate extends IndexedInputGate {
public List<InputChannelInfo> getUnfinishedChannels() {
return Collections.emptyList();
}
-
- @Override
- public void setCheckpointingDuringRecoveryEnabled(boolean enabled) {}
-
- @Override
- public boolean isCheckpointingDuringRecoveryEnabled() {
- return false;
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlignedCheckpointsMassiveRandomTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlignedCheckpointsMassiveRandomTest.java
index e27f6b28937..ad792d221c2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlignedCheckpointsMassiveRandomTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlignedCheckpointsMassiveRandomTest.java
@@ -268,11 +268,6 @@ class AlignedCheckpointsMassiveRandomTest {
return CompletableFuture.completedFuture(null);
}
- @Override
- public CompletableFuture<Void> getBufferFilteringCompleteFuture() {
- return CompletableFuture.completedFuture(null);
- }
-
@Override
public void finishReadRecoveredState() {}
@@ -291,13 +286,5 @@ class AlignedCheckpointsMassiveRandomTest {
public List<InputChannelInfo> getUnfinishedChannels() {
return Collections.emptyList();
}
-
- @Override
- public void setCheckpointingDuringRecoveryEnabled(boolean enabled) {}
-
- @Override
- public boolean isCheckpointingDuringRecoveryEnabled() {
- return false;
- }
}
}