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 f615ff7feea9b43bbed5daec908c71d39c459bb0 Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 00:24:01 2026 +0200 [FLINK-39521][network] Add RecoverableInputChannel contract and recovery sentinels Introduce the push-based recovery contract and its sentinels: - RecoverableInputChannel (@Internal): getChannelInfo(), onRecoveredStateBuffer(Buffer), finishRecoveredBufferDelivery(), insertRecoveryCheckpointBarrierIfInRecovery(long), requestRecoveryBufferBlocking(), onRecoveredStateConsumed(). - EndOfFetchedChannelStateEvent: singleton RuntimeEvent tail sentinel of the recovered-buffer stream; reflective write()/read() throw; deliberately distinct from EndOfInputChannelStateEvent. - RecoveryCheckpointBarrier (checkpoint.channel): per-checkpoint sentinel carried inside the recovery queue; write()/read() throw. - EventSerializer: type tags 13 (RECOVERY_CHECKPOINT_BARRIER_EVENT) and 14 (END_OF_FETCHED_CHANNEL_STATE_EVENT). Recovery completion is unified on the InputChannel base class: the new abstract getStateConsumedFuture() completes once a channel has consumed all of its recovered state (or never had any / was released). It is abstract rather than defaulted so a future subclass must state its recovery semantics explicitly. RecoveredInputChannel's package-private getter becomes a public override (field tightened to CompletableFuture<Void>); UnknownInputChannel returns a completed future (a channel with persisted state is always a RecoveredInputChannel at restore time). Local/RemoteInputChannel temporarily return a completed future until their push-based recovery state lands in the next commits. TestInputChannel implements the new contract as a spying test double. --- .../channel/RecoveryCheckpointBarrier.java | 69 ++++++++++++++++++++ .../network/api/serialization/EventSerializer.java | 20 ++++++ .../consumer/EndOfFetchedChannelStateEvent.java | 75 ++++++++++++++++++++++ .../network/partition/consumer/InputChannel.java | 8 +++ .../partition/consumer/LocalInputChannel.java | 7 ++ .../consumer/RecoverableInputChannel.java | 63 ++++++++++++++++++ .../partition/consumer/RecoveredInputChannel.java | 5 +- .../partition/consumer/RemoteInputChannel.java | 7 ++ .../partition/consumer/UnknownInputChannel.java | 7 ++ .../channel/RecoveryCheckpointBarrierTest.java | 53 +++++++++++++++ .../partition/consumer/InputChannelTest.java | 6 ++ .../partition/consumer/TestInputChannel.java | 49 +++++++++++++- 12 files changed, 366 insertions(+), 3 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveryCheckpointBarrier.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveryCheckpointBarrier.java new file mode 100644 index 00000000000..984bfd42312 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveryCheckpointBarrier.java @@ -0,0 +1,69 @@ +/* + * 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.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; +import org.apache.flink.runtime.event.RuntimeEvent; + +/** Task-local event marking the recovery-state cut for a recovery checkpoint. */ +@Internal +public final class RecoveryCheckpointBarrier extends RuntimeEvent { + + private final long checkpointId; + + public RecoveryCheckpointBarrier(long checkpointId) { + this.checkpointId = checkpointId; + } + + public long getCheckpointId() { + return checkpointId; + } + + @Override + public void write(DataOutputView out) { + throw new UnsupportedOperationException( + "RecoveryCheckpointBarrier must be serialized via EventSerializer's dedicated" + + " type-tag path, not reflective write()."); + } + + @Override + public void read(DataInputView in) { + throw new UnsupportedOperationException( + "RecoveryCheckpointBarrier must be deserialized via EventSerializer's dedicated" + + " type-tag path, not reflective read()."); + } + + @Override + public int hashCode() { + return Long.hashCode(checkpointId); + } + + @Override + public boolean equals(Object other) { + return other != null + && other.getClass() == RecoveryCheckpointBarrier.class + && ((RecoveryCheckpointBarrier) other).checkpointId == this.checkpointId; + } + + @Override + public String toString() { + return "RecoveryCheckpointBarrier(" + checkpointId + ")"; + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java index e73d9168cb2..5711d1640f9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java @@ -27,6 +27,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointOptions; import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.checkpoint.SavepointType; import org.apache.flink.runtime.checkpoint.SnapshotType; +import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointBarrier; import org.apache.flink.runtime.event.AbstractEvent; import org.apache.flink.runtime.event.WatermarkEvent; import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker; @@ -43,6 +44,7 @@ import org.apache.flink.runtime.io.network.buffer.Buffer; import org.apache.flink.runtime.io.network.buffer.BufferConsumer; 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.partition.consumer.EndOfFetchedChannelStateEvent; import org.apache.flink.runtime.io.network.partition.consumer.EndOfInputChannelStateEvent; import org.apache.flink.runtime.io.network.partition.consumer.EndOfOutputChannelStateEvent; import org.apache.flink.runtime.state.CheckpointStorageLocationReference; @@ -87,6 +89,10 @@ public class EventSerializer { private static final int END_OF_INPUT_CHANNEL_STATE_EVENT = 12; + private static final int RECOVERY_CHECKPOINT_BARRIER_EVENT = 13; + + private static final int END_OF_FETCHED_CHANNEL_STATE_EVENT = 14; + private static final byte CHECKPOINT_TYPE_CHECKPOINT = 0; private static final byte CHECKPOINT_TYPE_SAVEPOINT = 1; @@ -116,6 +122,8 @@ public class EventSerializer { return ByteBuffer.wrap(new byte[] {0, 0, 0, END_OF_OUTPUT_CHANNEL_STATE_EVENT}); } else if (eventClass == EndOfInputChannelStateEvent.class) { return ByteBuffer.wrap(new byte[] {0, 0, 0, END_OF_INPUT_CHANNEL_STATE_EVENT}); + } else if (eventClass == EndOfFetchedChannelStateEvent.class) { + return ByteBuffer.wrap(new byte[] {0, 0, 0, END_OF_FETCHED_CHANNEL_STATE_EVENT}); } else if (eventClass == EndOfData.class) { return ByteBuffer.wrap( new byte[] { @@ -162,6 +170,13 @@ public class EventSerializer { buf.putInt(0, RECOVERY_METADATA); buf.putInt(4, recoveryMetadata.getFinalBufferSubpartitionId()); return buf; + } else if (eventClass == RecoveryCheckpointBarrier.class) { + RecoveryCheckpointBarrier barrier = (RecoveryCheckpointBarrier) event; + + ByteBuffer buf = ByteBuffer.allocate(12); + buf.putInt(0, RECOVERY_CHECKPOINT_BARRIER_EVENT); + buf.putLong(4, barrier.getCheckpointId()); + return buf; } else if (eventClass == WatermarkEvent.class) { try { final DataOutputSerializer serializer = new DataOutputSerializer(128); @@ -206,6 +221,8 @@ public class EventSerializer { return EndOfOutputChannelStateEvent.INSTANCE; } else if (type == END_OF_INPUT_CHANNEL_STATE_EVENT) { return EndOfInputChannelStateEvent.INSTANCE; + } else if (type == END_OF_FETCHED_CHANNEL_STATE_EVENT) { + return EndOfFetchedChannelStateEvent.INSTANCE; } else if (type == END_OF_USER_RECORDS_EVENT) { return new EndOfData(StopMode.values()[buffer.get()]); } else if (type == CANCEL_CHECKPOINT_MARKER_EVENT) { @@ -222,6 +239,9 @@ public class EventSerializer { } else if (type == RECOVERY_METADATA) { int subpartitionId = buffer.getInt(); return new RecoveryMetadata(subpartitionId); + } else if (type == RECOVERY_CHECKPOINT_BARRIER_EVENT) { + long checkpointId = buffer.getLong(); + return new RecoveryCheckpointBarrier(checkpointId); } else if (type == GENERALIZED_WATERMARK_EVENT) { final DataInputDeserializer deserializer = new DataInputDeserializer(buffer); WatermarkEvent watermarkEvent = new WatermarkEvent(); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/EndOfFetchedChannelStateEvent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/EndOfFetchedChannelStateEvent.java new file mode 100644 index 00000000000..9900a0cdc66 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/EndOfFetchedChannelStateEvent.java @@ -0,0 +1,75 @@ +/* + * 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.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; +import org.apache.flink.runtime.event.RuntimeEvent; + +/** + * Marks the tail of recovered buffers that the spill drain pushed into a {@link + * RecoverableInputChannel}. The consume path polls this sentinel to learn the exact moment all + * recovered buffers have been consumed; it is never delivered to the operator. It is distinct from + * {@link EndOfInputChannelStateEvent} (which terminates the {@link RecoveredInputChannel} read + * stream) so the two recovery handoffs cannot be confused. + */ +public class EndOfFetchedChannelStateEvent extends RuntimeEvent { + + /** The singleton instance of this event. */ + public static final EndOfFetchedChannelStateEvent INSTANCE = + new EndOfFetchedChannelStateEvent(); + + // ------------------------------------------------------------------------ + + // not instantiable + private EndOfFetchedChannelStateEvent() {} + + // ------------------------------------------------------------------------ + + @Override + public void write(DataOutputView out) { + throw new UnsupportedOperationException( + "EndOfFetchedChannelStateEvent must be serialized via EventSerializer's dedicated" + + " type-tag path, not reflective write()."); + } + + @Override + public void read(DataInputView in) { + throw new UnsupportedOperationException( + "EndOfFetchedChannelStateEvent must be deserialized via EventSerializer's dedicated" + + " type-tag path, not reflective read()."); + } + + // ------------------------------------------------------------------------ + + @Override + public int hashCode() { + return 20250814; + } + + @Override + public boolean equals(Object obj) { + return obj != null && obj.getClass() == EndOfFetchedChannelStateEvent.class; + } + + @Override + public String toString() { + return getClass().getSimpleName(); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java index e2969c675ec..d96f99e40cb 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannel.java @@ -33,6 +33,7 @@ import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView; import java.io.IOException; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicReference; import static org.apache.flink.util.Preconditions.checkArgument; @@ -146,6 +147,13 @@ public abstract class InputChannel { return consumedSubpartitionIndexSet; } + /** + * Completes once this channel has consumed all of its recovered state: it never had state to + * recover, all recovered data was consumed, or the channel was released. Implementations that + * never participate in recovery must return an already completed future. + */ + public abstract CompletableFuture<Void> getStateConsumedFuture(); + /** * After sending a {@link org.apache.flink.runtime.io.network.api.CheckpointBarrier} of * exactly-once mode, the upstream will be blocked and become unavailable. This method tries to diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java index 001933a04c8..33338000ec8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java @@ -38,6 +38,7 @@ import org.apache.flink.runtime.io.network.partition.ResultPartitionManager; import org.apache.flink.runtime.io.network.partition.ResultSubpartition.BufferAndBacklog; import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet; import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView; +import org.apache.flink.util.concurrent.FutureUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -52,6 +53,7 @@ import java.util.List; import java.util.Optional; import java.util.Timer; import java.util.TimerTask; +import java.util.concurrent.CompletableFuture; import static org.apache.flink.util.Preconditions.checkNotNull; import static org.apache.flink.util.Preconditions.checkState; @@ -165,6 +167,11 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit channelStatePersister.startPersisting(barrier.getId(), inflightBuffers); } + @Override + public CompletableFuture<Void> getStateConsumedFuture() { + return FutureUtils.completedVoidFuture(); + } + public void checkpointStopped(long checkpointId) { channelStatePersister.stopPersisting(checkpointId); } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoverableInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoverableInputChannel.java new file mode 100644 index 00000000000..40a4ab27a09 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoverableInputChannel.java @@ -0,0 +1,63 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; +import org.apache.flink.runtime.io.network.buffer.Buffer; + +import java.io.IOException; + +/** Physical input channel that can receive recovered buffers pushed by the spill drain. */ +@Internal +public interface RecoverableInputChannel { + + InputChannelInfo getChannelInfo(); + + /** + * Appends a recovered buffer or recovery-checkpoint sentinel. Released channels recycle the + * buffer silently. + */ + void onRecoveredStateBuffer(Buffer buffer); + + /** + * Marks producer-side recovery delivery complete. Implementations wait for upstream readiness + * before flipping this state so channels without spill entries still observe the same handoff. + */ + void finishRecoveredBufferDelivery() throws IOException, InterruptedException; + + /** + * Inserts a {@code RecoveryCheckpointBarrier} for {@code checkpointId} into this channel's + * recovery queue if the channel is still in recovery. + */ + void insertRecoveryCheckpointBarrierIfInRecovery(long checkpointId) throws IOException; + + /** + * Blocks until a buffer is available from this channel's own buffer pool. Implementations must + * first await upstream readiness and must be invoked outside the drainer lock. + */ + Buffer requestRecoveryBufferBlocking() throws InterruptedException, IOException; + + /** + * Invoked by the consume path the moment it polls the {@code EndOfFetchedChannelStateEvent} + * sentinel, i.e. once all recovered buffers have been consumed. Implementations flip out of + * recovery, release any upstream events held back during recovery, and reopen the upstream so + * live data may flow again. + */ + void onRecoveredStateConsumed() 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 ed3bd0822a5..147848c7de6 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 @@ -59,7 +59,7 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan private static final Logger LOG = LoggerFactory.getLogger(RecoveredInputChannel.class); private final ArrayDeque<Buffer> receivedBuffers = new ArrayDeque<>(); - private final CompletableFuture<?> stateConsumedFuture = new CompletableFuture<>(); + private final CompletableFuture<Void> stateConsumedFuture = new CompletableFuture<>(); protected final BufferManager bufferManager; /** @@ -159,7 +159,8 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan return bufferFilteringCompleteFuture; } - CompletableFuture<?> getStateConsumedFuture() { + @Override + public CompletableFuture<Void> getStateConsumedFuture() { return stateConsumedFuture; } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java index 227bc44ca05..001fd45a58a 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java @@ -46,6 +46,7 @@ import org.apache.flink.runtime.io.network.partition.PrioritizedDeque; import org.apache.flink.runtime.io.network.partition.ResultPartitionID; import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet; import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.concurrent.FutureUtils; import org.apache.flink.shaded.guava33.com.google.common.collect.Iterators; @@ -62,6 +63,7 @@ import java.util.Iterator; import java.util.List; import java.util.Optional; import java.util.OptionalLong; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; @@ -740,6 +742,11 @@ public class RemoteInputChannel extends InputChannel { } } + @Override + public CompletableFuture<Void> getStateConsumedFuture() { + return FutureUtils.completedVoidFuture(); + } + public void checkpointStopped(long checkpointId) { synchronized (receivedBuffers) { channelStatePersister.stopPersisting(checkpointId); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java index 15182cedadb..ad59dd38db8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/UnknownInputChannel.java @@ -31,12 +31,14 @@ 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 org.apache.flink.util.Preconditions; +import org.apache.flink.util.concurrent.FutureUtils; import javax.annotation.Nullable; import java.io.IOException; import java.util.ArrayDeque; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_DECLINED_TASK_NOT_READY; import static org.apache.flink.util.Preconditions.checkNotNull; @@ -100,6 +102,11 @@ class UnknownInputChannel extends InputChannel implements ChannelStateHolder { this.networkBuffersPerChannel = networkBuffersPerChannel; } + @Override + public CompletableFuture<Void> getStateConsumedFuture() { + return FutureUtils.completedVoidFuture(); + } + @Override public void resumeConsumption() { throw new UnsupportedOperationException("UnknownInputChannel should never be blocked."); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveryCheckpointBarrierTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveryCheckpointBarrierTest.java new file mode 100644 index 00000000000..0dd2c18286a --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveryCheckpointBarrierTest.java @@ -0,0 +1,53 @@ +/* + * 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.core.memory.DataOutputSerializer; +import org.apache.flink.runtime.io.network.api.serialization.EventSerializer; +import org.apache.flink.runtime.io.network.buffer.Buffer; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link RecoveryCheckpointBarrier}. */ +class RecoveryCheckpointBarrierTest { + + @Test + void testReflectiveWriteIsUnsupported() { + RecoveryCheckpointBarrier barrier = new RecoveryCheckpointBarrier(1L); + + assertThatThrownBy(() -> barrier.write(new DataOutputSerializer(Long.BYTES))) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("dedicated type-tag path"); + } + + @Test + void testEventSerializerHandlesRecoveryCheckpointBarrier() throws Exception { + long checkpointId = 123L; + + Buffer buffer = + EventSerializer.toBuffer(new RecoveryCheckpointBarrier(checkpointId), false); + Object deserialized = + EventSerializer.fromBuffer( + buffer, RecoveryCheckpointBarrier.class.getClassLoader()); + + assertThat(deserialized).isEqualTo(new RecoveryCheckpointBarrier(checkpointId)); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelTest.java index 866636a3061..a6ade5890d1 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/InputChannelTest.java @@ -27,6 +27,7 @@ import org.junit.jupiter.api.Test; import java.io.IOException; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; @@ -145,6 +146,11 @@ class InputChannelTest { @Override public void resumeConsumption() {} + @Override + public CompletableFuture<Void> getStateConsumedFuture() { + return CompletableFuture.completedFuture(null); + } + @Override public void acknowledgeAllRecordsProcessed() throws IOException {} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java index 14a3654a666..8fd0a57f93c 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/TestInputChannel.java @@ -19,6 +19,7 @@ package org.apache.flink.runtime.io.network.partition.consumer; import org.apache.flink.metrics.SimpleCounter; +import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointBarrier; import org.apache.flink.runtime.event.TaskEvent; import org.apache.flink.runtime.io.network.api.EndOfData; import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent; @@ -31,8 +32,10 @@ import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet; import javax.annotation.Nullable; import java.io.IOException; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collection; +import java.util.Deque; import java.util.Optional; import java.util.Queue; import java.util.concurrent.CompletableFuture; @@ -45,7 +48,7 @@ import static org.apache.flink.util.Preconditions.checkState; import static org.assertj.core.api.Assertions.assertThat; /** A mocked input channel. */ -public class TestInputChannel extends InputChannel { +public class TestInputChannel extends InputChannel implements RecoverableInputChannel { private final Queue<BufferAndAvailabilityProvider> buffers = new ConcurrentLinkedQueue<>(); @@ -259,6 +262,50 @@ public class TestInputChannel extends InputChannel { requiredSegmentIdFuture.complete(segmentId); } + private final Deque<Buffer> recoveredBuffersSpy = new ArrayDeque<>(); + private boolean finishRecoveredBufferDeliveryCalled = false; + + @Override + public void onRecoveredStateBuffer(Buffer buffer) { + recoveredBuffersSpy.add(buffer); + } + + @Override + public void finishRecoveredBufferDelivery() { + finishRecoveredBufferDeliveryCalled = true; + } + + @Override + public void insertRecoveryCheckpointBarrierIfInRecovery(long checkpointId) throws IOException { + if (!finishRecoveredBufferDeliveryCalled || !recoveredBuffersSpy.isEmpty()) { + recoveredBuffersSpy.add( + EventSerializer.toBuffer(new RecoveryCheckpointBarrier(checkpointId), false)); + } + } + + @Override + public Buffer requestRecoveryBufferBlocking() { + throw new UnsupportedOperationException("TestInputChannel does not back recovery drain"); + } + + @Override + public void onRecoveredStateConsumed() { + // No-op in this test stub. + } + + @Override + public CompletableFuture<Void> getStateConsumedFuture() { + return CompletableFuture.completedFuture(null); + } + + public Deque<Buffer> getRecoveredBuffersSpy() { + return recoveredBuffersSpy; + } + + public boolean isFinishRecoveredBufferDeliveryCalled() { + return finishRecoveredBufferDeliveryCalled; + } + public void assertReturnedEventsAreRecycled() { assertReturnedBuffersAreRecycled(false, true); }
