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