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 e480c74453db5cc880e8ebd3eb62e5c7d3aa86ee Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 00:29:49 2026 +0200 [FLINK-39521][network] Convert recovered channels via the push interface; thread needsRecovery through gates RecoveredInputChannel#toInputChannel(boolean needsRecovery) replaces the no-arg variant; constructor migration of recovered buffers is retired: - needsRecovery=false: checkState(receivedBuffers.isEmpty()) + inputChannel.setup() (setup moves from RemoteRecoveredInputChannel into the conversion) + checkpointStopped(...). Flag-off conversion always happens after full consumption, so default behavior is unchanged. - needsRecovery=true (first used by the StreamTask recovery rework PR): create the physical channel in recovery state, then synchronously push every queued recovered data buffer via onRecoveredStateBuffer and append the EndOfFetchedChannelStateEvent sentinel (the legacy EndOfInputChannelStateEvent in the queue is dropped in translation). This is the simple in-memory drain -- FLINK-38544 transitional code, replaced by the disk drain when the spilling backend lands. The sentinel is appended directly rather than via finishRecoveredBufferDelivery(), which waits for upstream readiness that cannot arrive while the mailbox thread is still converting channels. - abstract toInputChannelInternal(ArrayDeque<Buffer>) becomes toInputChannelInternal(boolean). Gates thread the flag through: InputGate gains a concrete requestPartitions(boolean) (default overload ignores the flag); SingleInputGate implements requestPartitions(boolean) and convertRecoveredInputChannels(boolean) (no-arg kept as a @VisibleForTesting shim); getStateConsumedFuture() aggregation is unified on InputChannel#getStateConsumedFuture across all channels; UnionInputGate fans the flag out to member gates; InputGateWithMetrics delegates. Deliberate consequence: the base v1 flag-on StreamTask still converts at filtering-complete with unconsumed buffers, so the empty-queue checkState fires and the flag-on path is degraded until the StreamTask recovery rework lands -- covered by the randomization pin in the first commit of this PR. Default (flag-off) configuration is unaffected. Tests: RecoveredInputChannelTest is rewritten around the new conversion semantics (reject-while-unconsumed, empty-queue assert, and a transitional push-conversion case); Local/RemoteRecoveredInputChannelTest cover the empty-queue assert of the subclasses; UnionInputGateTest's helper adapts to the new toInputChannelInternal signature. --- .../io/network/partition/consumer/InputGate.java | 9 ++ .../consumer/LocalRecoveredInputChannel.java | 9 +- .../partition/consumer/RecoveredInputChannel.java | 61 ++++++--- .../consumer/RemoteRecoveredInputChannel.java | 10 +- .../partition/consumer/SingleInputGate.java | 25 ++-- .../network/partition/consumer/UnionInputGate.java | 7 +- .../runtime/taskmanager/InputGateWithMetrics.java | 5 + .../consumer/LocalRecoveredInputChannelTest.java | 51 ++++++++ .../consumer/RecoveredInputChannelTest.java | 137 +++++++++------------ .../consumer/RemoteRecoveredInputChannelTest.java | 51 ++++++++ .../partition/consumer/UnionInputGateTest.java | 4 +- 11 files changed, 245 insertions(+), 124 deletions(-) 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 8c5b02fb565..2348c33ae98 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 @@ -198,6 +198,15 @@ public abstract class InputGate public abstract void requestPartitions() throws IOException; + /** + * Requests the partitions. {@code needsRecovery} controls whether converted physical channels + * start in recovery (i.e. with no credit / no floating buffers until the recovered channel + * state has been drained). The default implementation ignores the flag. + */ + public void requestPartitions(boolean needsRecovery) throws IOException { + requestPartitions(); + } + public abstract CompletableFuture<Void> getStateConsumedFuture(); /** diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannel.java index 8ec00582215..de058bd3723 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannel.java @@ -19,14 +19,11 @@ package org.apache.flink.runtime.io.network.partition.consumer; import org.apache.flink.runtime.io.network.TaskEventPublisher; -import org.apache.flink.runtime.io.network.buffer.Buffer; import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics; import org.apache.flink.runtime.io.network.partition.ResultPartitionID; import org.apache.flink.runtime.io.network.partition.ResultPartitionManager; import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet; -import java.util.ArrayDeque; - import static org.apache.flink.util.Preconditions.checkNotNull; /** @@ -64,9 +61,7 @@ public class LocalRecoveredInputChannel extends RecoveredInputChannel { } @Override - protected InputChannel toInputChannelInternal(ArrayDeque<Buffer> remainingBuffers) { - // remainingBuffers is unused: with the flag off the queue is always empty at conversion, - // and constructor migration is retired by the conversion rework later in this PR. + protected InputChannel toInputChannelInternal(boolean needsRecovery) { return new LocalInputChannel( inputGate, getChannelIndex(), @@ -80,6 +75,6 @@ public class LocalRecoveredInputChannel extends RecoveredInputChannel { numBuffersIn, channelStateWriter, networkBuffersPerChannel, - false); + needsRecovery); } } 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 147848c7de6..ec878b69c62 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 @@ -115,23 +115,53 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan this.channelStateWriter = checkNotNull(channelStateWriter); } - public final InputChannel toInputChannel() throws IOException { - Preconditions.checkState( - bufferFilteringCompleteFuture.isDone(), "buffer filtering is not complete"); - if (!inputGate.isCheckpointingDuringRecoveryEnabled()) { - Preconditions.checkState( - stateConsumedFuture.isDone(), "recovered state is not fully consumed"); + public final InputChannel toInputChannel(boolean needsRecovery) throws IOException { + if (needsRecovery) { + return toInputChannelInRecovery(); + } + synchronized (receivedBuffers) { + Preconditions.checkState(receivedBuffers.isEmpty(), "Received buffer should be empty."); } - // Extract remaining buffers before conversion. - // These buffers have been filtered but not yet consumed by the Task. - final ArrayDeque<Buffer> remainingBuffers; + final InputChannel inputChannel = toInputChannelInternal(needsRecovery); + inputChannel.setup(); + inputChannel.checkpointStopped(lastStoppedCheckpointId); + return inputChannel; + } + + /** + * FLINK-38544 transitional: removed when the spilling backend lands. Creates the physical + * channel in recovery state and synchronously hands every queued recovered buffer over through + * the push interface. The legacy {@link EndOfInputChannelStateEvent} in the queue is dropped in + * translation; the {@link EndOfFetchedChannelStateEvent} sentinel takes its place. The sentinel + * is appended directly instead of via {@link + * RecoverableInputChannel#finishRecoveredBufferDelivery()} because that method waits for + * upstream readiness, which cannot happen while the mailbox thread is still converting channels + * (partitions are requested only after conversion). + */ + private InputChannel toInputChannelInRecovery() throws IOException { + final Buffer[] remainingBuffers; synchronized (receivedBuffers) { - remainingBuffers = new ArrayDeque<>(receivedBuffers); + remainingBuffers = receivedBuffers.toArray(new Buffer[0]); receivedBuffers.clear(); } - final InputChannel inputChannel = toInputChannelInternal(remainingBuffers); + final InputChannel inputChannel = toInputChannelInternal(true); + inputChannel.setup(); + final RecoverableInputChannel recoverableChannel = (RecoverableInputChannel) inputChannel; + for (int i = 0; i < remainingBuffers.length; i++) { + final Buffer buffer = remainingBuffers[i]; + if (isEndOfInputChannelStateEvent(buffer)) { + Preconditions.checkState( + i == remainingBuffers.length - 1, + "EndOfInputChannelStateEvent must be the last recovered buffer."); + buffer.recycleBuffer(); + } else { + recoverableChannel.onRecoveredStateBuffer(buffer); + } + } + recoverableChannel.onRecoveredStateBuffer( + EventSerializer.toBuffer(EndOfFetchedChannelStateEvent.INSTANCE, false)); inputChannel.checkpointStopped(lastStoppedCheckpointId); return inputChannel; } @@ -142,13 +172,10 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan } /** - * Creates the physical InputChannel from this recovered channel. - * - * @param remainingBuffers buffers that have been filtered but not yet consumed by the Task. - * These buffers will be migrated to the new physical channel. - * @return the physical InputChannel (LocalInputChannel or RemoteInputChannel) + * Creates the physical {@link InputChannel}; {@code needsRecovery} controls whether it starts + * in recovery. */ - protected abstract InputChannel toInputChannelInternal(ArrayDeque<Buffer> remainingBuffers) + protected abstract InputChannel toInputChannelInternal(boolean needsRecovery) throws IOException; /** diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannel.java index 6b0279ad3e2..b76aa347fe6 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannel.java @@ -20,13 +20,11 @@ package org.apache.flink.runtime.io.network.partition.consumer; import org.apache.flink.runtime.io.network.ConnectionID; import org.apache.flink.runtime.io.network.ConnectionManager; -import org.apache.flink.runtime.io.network.buffer.Buffer; import org.apache.flink.runtime.io.network.metrics.InputChannelMetrics; import org.apache.flink.runtime.io.network.partition.ResultPartitionID; import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet; import java.io.IOException; -import java.util.ArrayDeque; import static org.apache.flink.util.Preconditions.checkNotNull; @@ -68,10 +66,7 @@ public class RemoteRecoveredInputChannel extends RecoveredInputChannel { } @Override - protected InputChannel toInputChannelInternal(ArrayDeque<Buffer> remainingBuffers) - throws IOException { - // remainingBuffers is unused: with the flag off the queue is always empty at conversion, - // and constructor migration is retired by the conversion rework later in this PR. + protected InputChannel toInputChannelInternal(boolean needsRecovery) throws IOException { RemoteInputChannel remoteInputChannel = new RemoteInputChannel( inputGate, @@ -87,8 +82,7 @@ public class RemoteRecoveredInputChannel extends RecoveredInputChannel { numBytesIn, numBuffersIn, channelStateWriter, - false); - remoteInputChannel.setup(); + needsRecovery); return remoteInputChannel; } } 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 1e6b8168f34..4a7c9c6617e 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 @@ -323,9 +323,7 @@ public class SingleInputGate extends IndexedInputGate { synchronized (requestLock) { List<CompletableFuture<?>> futures = new ArrayList<>(numberOfInputChannels); for (InputChannel inputChannel : inputChannels()) { - if (inputChannel instanceof RecoveredInputChannel) { - futures.add(((RecoveredInputChannel) inputChannel).getStateConsumedFuture()); - } + futures.add(inputChannel.getStateConsumedFuture()); } return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); } @@ -358,6 +356,11 @@ public class SingleInputGate extends IndexedInputGate { @Override public void requestPartitions() { + requestPartitions(false); + } + + @Override + public void requestPartitions(boolean needsRecovery) { synchronized (requestLock) { if (!requestedPartitionsFlag) { if (closeFuture.isDone()) { @@ -376,7 +379,7 @@ public class SingleInputGate extends IndexedInputGate { numInputChannels, numberOfInputChannels)); } - convertRecoveredInputChannels(); + convertRecoveredInputChannels(needsRecovery); internalRequestPartitions(); } @@ -390,12 +393,18 @@ public class SingleInputGate extends IndexedInputGate { } } + @VisibleForTesting + public void convertRecoveredInputChannels() { + convertRecoveredInputChannels(false); + } + /** * Converts all {@link RecoveredInputChannel}s to their real channel types ({@link - * LocalInputChannel} or {@link RemoteInputChannel}). + * LocalInputChannel} or {@link RemoteInputChannel}). {@code needsRecovery} controls whether the + * converted physical channels start in recovery. */ @VisibleForTesting - public void convertRecoveredInputChannels() { + public void convertRecoveredInputChannels(boolean needsRecovery) { LOG.debug("Converting recovered input channels ({} channels)", getNumberOfInputChannels()); for (Map<InputChannelInfo, InputChannel> inputChannelsForCurrentPartition : inputChannels.values()) { @@ -413,7 +422,7 @@ public class SingleInputGate extends IndexedInputGate { // order with onRecoveredStateBuffer() which acquires receivedBuffers // first and then inputChannelsWithData. InputChannel realInputChannel = - ((RecoveredInputChannel) inputChannel).toInputChannel(); + ((RecoveredInputChannel) inputChannel).toInputChannel(needsRecovery); inputChannel.releaseAllResources(); int buffersInUseCount = realInputChannel.getBuffersInUseCount(); @@ -962,7 +971,7 @@ public class SingleInputGate extends IndexedInputGate { // Firstly, read the buffers from the recovered channel if (inputChannel instanceof RecoveredInputChannel && !inputChannel.isReleased()) { Optional<Buffer> buffer = readBufferFromInputChannel(inputChannel); - if (!((RecoveredInputChannel) inputChannel).getStateConsumedFuture().isDone()) { + if (!inputChannel.getStateConsumedFuture().isDone()) { return buffer; } } 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 8d1ad255377..37aee532ae5 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 @@ -368,8 +368,13 @@ public class UnionInputGate extends InputGate { @Override public void requestPartitions() throws IOException { + requestPartitions(false); + } + + @Override + public void requestPartitions(boolean needsRecovery) throws IOException { for (InputGate inputGate : inputGatesByGateIndex.values()) { - inputGate.requestPartitions(); + inputGate.requestPartitions(needsRecovery); } } 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 3eff82a6eb9..695d897c92f 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 @@ -135,6 +135,11 @@ public class InputGateWithMetrics extends IndexedInputGate { inputGate.requestPartitions(); } + @Override + public void requestPartitions(boolean needsRecovery) throws IOException { + inputGate.requestPartitions(needsRecovery); + } + @Override public void setChannelStateWriter(ChannelStateWriter channelStateWriter) { inputGate.setChannelStateWriter(channelStateWriter); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannelTest.java new file mode 100644 index 00000000000..fdacf6ec1ee --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannelTest.java @@ -0,0 +1,51 @@ +/* + * 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.io.network.partition.consumer; + +import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter; +import org.apache.flink.runtime.io.network.buffer.Buffer; +import org.apache.flink.runtime.io.network.util.TestBufferFactory; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class LocalRecoveredInputChannelTest { + + @Test + void testToInputChannelRequiresEmptyRecoveredBuffers() throws Exception { + SingleInputGate inputGate = new SingleInputGateBuilder().build(); + LocalRecoveredInputChannel recoveredChannel = + InputChannelBuilder.newBuilder() + .setStateWriter(ChannelStateWriter.NO_OP) + .buildLocalRecoveredChannel(inputGate); + + Buffer buffer = TestBufferFactory.createBuffer(11); + recoveredChannel.onRecoveredStateBuffer(buffer); + + try { + recoveredChannel.finishReadRecoveredState(); + assertThatThrownBy(() -> recoveredChannel.toInputChannel(false)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Received buffer should be empty"); + } finally { + recoveredChannel.releaseAllResources(); + } + } +} 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 f40fd09702e..87a6c0466de 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 @@ -22,14 +22,15 @@ import org.apache.flink.metrics.SimpleCounter; import org.apache.flink.runtime.checkpoint.CheckpointException; import org.apache.flink.runtime.checkpoint.CheckpointType; 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.BufferBuilderTestUtils; import org.apache.flink.runtime.io.network.partition.ResultPartitionID; import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet; import org.junit.jupiter.api.Test; import java.io.IOException; -import java.util.ArrayDeque; import static org.apache.flink.runtime.checkpoint.CheckpointOptions.unaligned; import static org.apache.flink.runtime.state.CheckpointStorageLocationReference.getDefault; @@ -39,16 +40,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link RecoveredInputChannel}. */ class RecoveredInputChannelTest { - @Test - void testConversionOnlyPossibleAfterBufferFilteringComplete() { - // toInputChannel() always checks bufferFilteringCompleteFuture regardless of config - for (boolean configEnabled : new boolean[] {true, false}) { - assertThatThrownBy(() -> buildChannel(configEnabled).toInputChannel()) - .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("buffer filtering is not complete"); - } - } - @Test void testRequestPartitionsImpossible() { assertThatThrownBy(() -> buildChannel(false).requestSubpartitions()) @@ -71,93 +62,83 @@ class RecoveredInputChannelTest { } @Test - void testToInputChannelAllowedWhenBufferFilteringCompleteAndConfigEnabled() throws IOException { - // When config is enabled, conversion is allowed when bufferFilteringCompleteFuture is done - TestableRecoveredInputChannel channel = buildTestableChannel(true); - - // Initially, conversion should fail - assertThatThrownBy(() -> channel.toInputChannel()) - .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("buffer filtering is not complete"); - - // After finishReadRecoveredState(), bufferFilteringCompleteFuture should be done - channel.finishReadRecoveredState(); - assertThat(channel.getBufferFilteringCompleteFuture()).isDone(); - assertThat(channel.getStateConsumedFuture()).isNotDone(); - - // Conversion should now succeed (no exception) - InputChannel converted = channel.toInputChannel(); - assertThat(converted).isNotNull(); - } - - @Test - void testToInputChannelAllowedWhenStateConsumedAndConfigDisabled() throws IOException { - // When config is disabled, conversion requires both bufferFilteringCompleteFuture - // and stateConsumedFuture to be done + void testToInputChannelRejectedWhileRecoveredStateUnconsumed() throws IOException { + // Conversion is rejected while recovered state is still queued: finishReadRecoveredState() + // enqueues the EndOfInputChannelStateEvent sentinel, so receivedBuffers is non-empty until + // it is consumed. The empty-queue check thus also guarantees stateConsumedFuture is done. TestableRecoveredInputChannel channel = buildTestableChannel(false); - // Initially, conversion should fail (buffer filtering not complete) - assertThatThrownBy(() -> channel.toInputChannel()) - .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("buffer filtering is not complete"); - - // After finishReadRecoveredState(), bufferFilteringCompleteFuture is done - // but stateConsumedFuture is not channel.finishReadRecoveredState(); - assertThat(channel.getBufferFilteringCompleteFuture()).isDone(); assertThat(channel.getStateConsumedFuture()).isNotDone(); - // Conversion should still fail because stateConsumedFuture is not done - assertThatThrownBy(() -> channel.toInputChannel()) + // Conversion fails because the sentinel is still queued. + assertThatThrownBy(() -> channel.toInputChannel(false)) .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("recovered state is not fully consumed"); + .hasMessageContaining("Received buffer should be empty"); - // Consume the EndOfInputChannelStateEvent to complete stateConsumedFuture + // Consuming the EndOfInputChannelStateEvent should complete the future. + // getNextBuffer() returns empty when it encounters the event internally. assertThat(channel.getNextBuffer()).isNotPresent(); assertThat(channel.getStateConsumedFuture()).isDone(); // Now conversion should succeed - InputChannel converted = channel.toInputChannel(); + InputChannel converted = channel.toInputChannel(true); assertThat(converted).isNotNull(); } @Test - void testBufferFilteringCompleteFutureAlwaysCompletes() throws IOException { - // finishReadRecoveredState() unconditionally completes bufferFilteringCompleteFuture - for (boolean configEnabled : new boolean[] {true, false}) { - RecoveredInputChannel channel = buildChannel(configEnabled); - assertThat(channel.getBufferFilteringCompleteFuture()).isNotDone(); - channel.finishReadRecoveredState(); - assertThat(channel.getBufferFilteringCompleteFuture()).isDone(); - } + void testToInputChannelRequiresEmptyRecoveredBuffers() throws IOException { + TestableRecoveredInputChannel channel = buildTestableChannel(true); + + channel.onRecoveredStateBuffer(BufferBuilderTestUtils.buildSomeBuffer()); + channel.finishReadRecoveredState(); + + assertThatThrownBy(() -> channel.toInputChannel(false)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Received buffer should be empty"); } @Test - void testStateConsumedFutureCompletesAfterConsumingAllBuffers() throws IOException { - // This test verifies that stateConsumedFuture completes after consuming - // EndOfInputChannelStateEvent regardless of the config setting - for (boolean configEnabled : new boolean[] {true, false}) { - RecoveredInputChannel channel = buildChannel(configEnabled); + void testToInputChannelPushesQueuedBuffersWhenNeedsRecovery() throws IOException { + // FLINK-38544 transitional: removed when the spilling backend lands (recovered state then + // goes to disk, the queue is always empty at conversion, and toInputChannel(true) asserts + // emptiness instead of pushing). + TestableRecoveredInputChannel channel = buildTestableChannel(true); - assertThat(channel.getStateConsumedFuture()).isNotDone(); + channel.onRecoveredStateBuffer(BufferBuilderTestUtils.buildSomeBuffer(42)); + channel.finishReadRecoveredState(); - channel.finishReadRecoveredState(); - assertThat(channel.getStateConsumedFuture()).isNotDone(); + TestInputChannel converted = (TestInputChannel) channel.toInputChannel(true); + + // The queued data buffer is handed over through the push interface, the legacy + // EndOfInputChannelStateEvent is dropped in translation, and the + // EndOfFetchedChannelStateEvent sentinel is appended after the last recovered buffer. + assertThat(converted.getRecoveredBuffersSpy()).hasSize(2); + Buffer data = converted.getRecoveredBuffersSpy().pollFirst(); + assertThat(data.isBuffer()).isTrue(); + assertThat(data.getSize()).isEqualTo(42); + Buffer sentinel = converted.getRecoveredBuffersSpy().pollFirst(); + assertThat(sentinel.isBuffer()).isFalse(); + assertThat(EventSerializer.fromBuffer(sentinel, getClass().getClassLoader())) + .isInstanceOf(EndOfFetchedChannelStateEvent.class); + } - // Consuming the EndOfInputChannelStateEvent should complete the future. - // getNextBuffer() returns empty when it encounters the event internally. - assertThat(channel.getNextBuffer()).isNotPresent(); - assertThat(channel.getStateConsumedFuture()).isDone(); - } + @Test + void testStateConsumedFutureCompletesAfterLegacySentinelIsConsumed() throws IOException { + RecoveredInputChannel channel = buildChannel(false); + + assertThat(channel.getStateConsumedFuture()).isNotDone(); + + channel.finishReadRecoveredState(); + assertThat(channel.getStateConsumedFuture()).isNotDone(); + + assertThat(channel.getNextBuffer()).isNotPresent(); + assertThat(channel.getStateConsumedFuture()).isDone(); } private RecoveredInputChannel buildChannel(boolean checkpointingDuringRecoveryEnabled) { try { - SingleInputGate inputGate = - new SingleInputGateBuilder() - .setCheckpointingDuringRecoveryEnabled( - checkpointingDuringRecoveryEnabled) - .build(); + SingleInputGate inputGate = new SingleInputGateBuilder().build(); return new RecoveredInputChannel( inputGate, 0, @@ -169,7 +150,7 @@ class RecoveredInputChannelTest { new SimpleCounter(), 10) { @Override - protected InputChannel toInputChannelInternal(ArrayDeque<Buffer> remainingBuffers) { + protected InputChannel toInputChannelInternal(boolean needsRecovery) { throw new AssertionError("channel conversion succeeded"); } }; @@ -181,11 +162,7 @@ class RecoveredInputChannelTest { private TestableRecoveredInputChannel buildTestableChannel( boolean checkpointingDuringRecoveryEnabled) { try { - SingleInputGate inputGate = - new SingleInputGateBuilder() - .setCheckpointingDuringRecoveryEnabled( - checkpointingDuringRecoveryEnabled) - .build(); + SingleInputGate inputGate = new SingleInputGateBuilder().build(); return new TestableRecoveredInputChannel(inputGate); } catch (Exception e) { throw new AssertionError("channel creation failed", e); @@ -210,7 +187,7 @@ class RecoveredInputChannelTest { } @Override - protected InputChannel toInputChannelInternal(ArrayDeque<Buffer> remainingBuffers) { + protected InputChannel toInputChannelInternal(boolean needsRecovery) { return new TestInputChannel(inputGate, 0); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannelTest.java new file mode 100644 index 00000000000..7aa0461e0d9 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannelTest.java @@ -0,0 +1,51 @@ +/* + * 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.io.network.partition.consumer; + +import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter; +import org.apache.flink.runtime.io.network.buffer.Buffer; +import org.apache.flink.runtime.io.network.util.TestBufferFactory; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class RemoteRecoveredInputChannelTest { + + @Test + void testToInputChannelRequiresEmptyRecoveredBuffers() throws Exception { + SingleInputGate inputGate = new SingleInputGateBuilder().build(); + RemoteRecoveredInputChannel recoveredChannel = + InputChannelBuilder.newBuilder() + .setStateWriter(ChannelStateWriter.NO_OP) + .buildRemoteRecoveredChannel(inputGate); + + Buffer buffer = TestBufferFactory.createBuffer(13); + recoveredChannel.onRecoveredStateBuffer(buffer); + + try { + recoveredChannel.finishReadRecoveredState(); + assertThatThrownBy(() -> recoveredChannel.toInputChannel(false)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Received buffer should be empty"); + } finally { + recoveredChannel.releaseAllResources(); + } + } +} 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 1ed1a42a66e..a7325d15375 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 @@ -325,9 +325,7 @@ class UnionInputGateTest extends InputGateTestBase { new SimpleCounter(), 10) { @Override - protected InputChannel toInputChannelInternal( - java.util.ArrayDeque<org.apache.flink.runtime.io.network.buffer.Buffer> - remainingBuffers) { + protected InputChannel toInputChannelInternal(boolean needsRecovery) { throw new UnsupportedOperationException(); } };
