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 ed201228225fa51832734958a555594a21113544 Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 01:36:00 2026 +0200 [FLINK-39523] Thread the trigger through barrier-handler construction and the StreamTask lifecycle InputProcessorUtil.createCheckpointBarrierHandler: overload taking a RecoveryCheckpointTrigger (the old signature is preserved as a NO_OP delegator); SingleCheckpointBarrierHandler.unaligned/alternating gain the trigger param (aligned keeps the NO_OP default through ChannelState's 1-arg ctor); OneInputStreamTask/TwoInputStreamTask/MultipleInputStreamTask pass StreamTask.getRecoveryCheckpointTrigger(). (The channel-state writer is threaded through the same seams when the spilling backend lands.) StreamTask: recoveryCheckpointTrigger field (starts NOT_READY) + mailbox-thread-asserting getter lambda; all asynchronous mutations via setRecoveryCheckpointTrigger (a mailbox mail). Lifecycle inside recoverChannelsWithCheckpointing: NOT_READY before/during filtering (checkpoints declined as task-not-ready; transient, the coordinator retries) -> install the in-memory barrier-inserting trigger at conversion, composing on requestPartitions' returned List<RecoverableInputChannel> -> swap to NO_OP when completeAll(gates' stateConsumedFutures) completes (gate on the futures, not on push completion; the mailbox-mail gap is safe because a barrier-inserting trigger with no in-recovery channels behaves as NO_OP). recoverChannelsWithoutCheckpointing and the empty-input-gates short-circuit go straight to NO_OP. Also fix the transitional conversion push's sentinel timing: the EndOfFetchedChannelStateEvent sentinel is no longer appended inside toInputChannelInRecovery() (where it becomes consumable before the upstream connection exists, letting a spurious post-recovery poll hit a LocalInputChannel without a subpartition view: "Queried for a buffer before requesting the subpartition"). Instead the recovery chain appends it via finishRecoveredBufferDelivery() on the channelIOExecutor after partitions are requested -- the method waits for upstream readiness, restoring the invariant the disk drainer also relies on, and mirroring where the drainer will run. Tests: TestBarrierHandlerFactory adaptation; StreamTask trigger-lifecycle source-level invariants (transitional, replaced with the disk drainer wiring when the spilling backend lands); RecoveredInputChannelTest push-conversion case adapted to the sentinel timing. --- .../checkpoint/channel/FetchedChannelState.java | 67 +++++++++++ .../channel/FetchedChannelStateDrainer.java | 84 ++++++++++++++ .../channel/SequentialChannelStateReader.java | 12 +- .../channel/SequentialChannelStateReaderImpl.java | 47 ++++++-- .../partition/consumer/RecoveredInputChannel.java | 19 ++-- .../io/checkpointing/InputProcessorUtil.java | 32 +++++- .../SingleCheckpointBarrierHandler.java | 10 +- .../runtime/tasks/MultipleInputStreamTask.java | 3 +- .../runtime/tasks/OneInputStreamTask.java | 3 +- .../flink/streaming/runtime/tasks/StreamTask.java | 125 ++++++++++++++++++--- .../runtime/tasks/TwoInputStreamTask.java | 3 +- .../consumer/RecoveredInputChannelTest.java | 15 +-- .../checkpointing/TestBarrierHandlerFactory.java | 2 + 13 files changed, 367 insertions(+), 55 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java new file mode 100644 index 00000000000..6e28c525dbb --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java @@ -0,0 +1,67 @@ +/* + * 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.annotation.VisibleForTesting; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.file.Path; +import java.util.Collections; +import java.util.List; + +/** + * Sealed container for fetched recovered channel-state data. + * + * <p>FLINK-38544 transitional in-memory placeholder: the in-memory recovery backend keeps all + * recovered buffers inside the physical channels' own queues (pushed in one shot at conversion + * time), so there is no spill file and nothing to hand out or clean up here. This container carries + * only the "there is state to recover" signal that {@link + * SequentialChannelStateReader#readInputData} returns to the {@code StreamTask} recovery link; the + * lifecycle and file APIs are no-ops until the spilling backend lands and replaces this with a + * real, file-backed container. + */ +@Internal +public final class FetchedChannelState implements Closeable { + + private volatile boolean closed = false; + + FetchedChannelState() {} + + /** Returns the ordered list of spill file paths; empty for the in-memory backend. */ + public List<Path> files() { + return Collections.emptyList(); + } + + /** Acquires a lifecycle grant; no-op for the in-memory backend (no files to keep alive). */ + public void acquire() {} + + /** Releases a lifecycle grant; no-op for the in-memory backend (no files to delete). */ + public void release() throws IOException {} + + @Override + public void close() throws IOException { + closed = true; + } + + @VisibleForTesting + public boolean isClosed() { + return closed; + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainer.java new file mode 100644 index 00000000000..9d5a84d8330 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainer.java @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.runtime.checkpoint.channel; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.io.network.partition.consumer.RecoverableInputChannel; + +import java.io.Closeable; +import java.io.IOException; +import java.util.List; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * Drains a {@link FetchedChannelState} into the physical recovered channels and inserts {@link + * RecoveryCheckpointBarrier}s when a checkpoint fires during recovery. + * + * <p>FLINK-38544 transitional in-memory implementation: the in-memory recovery backend already + * pushed every recovered buffer into the physical channels' own queues at conversion time (via + * {@code requestPartitions(true)}), so draining is just appending the end-of-recovered-state + * sentinel to each channel, and there is never an undrained residue to snapshot — inserting the + * barrier into the in-recovery channels is enough. The disk-based drainer of the spilling backend + * replaces this, reading segments off spill files and returning a reader over the undrained slice. + */ +@Internal +public final class FetchedChannelStateDrainer implements RecoveryCheckpointTrigger, Closeable { + + private final FetchedChannelState channelState; + + private final List<RecoverableInputChannel> channels; + + public FetchedChannelStateDrainer( + FetchedChannelState channelState, List<RecoverableInputChannel> channels) { + this.channelState = checkNotNull(channelState); + this.channels = checkNotNull(channels); + } + + /** + * Appends the end-of-recovered-state sentinel to every converted channel. Each channel first + * waits for its upstream to be ready, so this must run on the channelIOExecutor rather than + * block the mailbox thread. Only once the sentinel is in place can the consume path flip the + * channel out of recovery, which guarantees live data is never polled before the upstream + * connection exists. + */ + public void drain() throws IOException, InterruptedException { + channelState.release(); + for (RecoverableInputChannel channel : channels) { + channel.finishRecoveredBufferDelivery(); + } + } + + /** + * Inserts a {@link RecoveryCheckpointBarrier} into every channel that is still in recovery, so + * that {@code checkpointStarted}'s in-recovery branch can persist exactly the pre-barrier + * recovered data. There is no snapshot side for the in-memory backend: everything a checkpoint + * must persist is already inside the channels' queues, so the snapshot is inherently empty. + */ + @Override + public void snapshotAndInsertBarriers(long checkpointId) throws IOException { + for (RecoverableInputChannel channel : channels) { + channel.insertRecoveryCheckpointBarrierIfInRecovery(checkpointId); + } + } + + @Override + public void close() throws IOException { + channelState.close(); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java index 547b60ef93a..f6b8decf09c 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java @@ -23,6 +23,7 @@ import org.apache.flink.runtime.io.network.partition.consumer.InputGate; import org.apache.flink.streaming.runtime.io.recovery.RecordFilterContext; import java.io.IOException; +import java.util.Optional; /** Reads channel state saved during checkpoint/savepoint. */ @Internal @@ -33,8 +34,11 @@ public interface SequentialChannelStateReader extends AutoCloseable { * * @param inputGates The input gates to recover state for. * @param filterContext The filter context containing input configs and rescaling info. + * @return the fetched recovered channel state if there is any state to recover, otherwise + * {@link Optional#empty()}. */ - void readInputData(InputGate[] inputGates, RecordFilterContext filterContext) + Optional<FetchedChannelState> readInputData( + InputGate[] inputGates, RecordFilterContext filterContext) throws IOException, InterruptedException; void readOutputData(ResultPartitionWriter[] writers, boolean notifyAndBlockOnCompletion) @@ -47,8 +51,10 @@ public interface SequentialChannelStateReader extends AutoCloseable { new SequentialChannelStateReader() { @Override - public void readInputData( - InputGate[] inputGates, RecordFilterContext filterContext) {} + public Optional<FetchedChannelState> readInputData( + InputGate[] inputGates, RecordFilterContext filterContext) { + return Optional.empty(); + } @Override public void readOutputData( diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java index 3ebad6eb3a2..e354d1ddf8b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java @@ -36,6 +36,7 @@ import java.util.Collection; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.function.Consumer; import java.util.function.Function; @@ -60,7 +61,8 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR } @Override - public void readInputData(InputGate[] inputGates, RecordFilterContext filterContext) + public Optional<FetchedChannelState> readInputData( + InputGate[] inputGates, RecordFilterContext filterContext) throws IOException, InterruptedException { // Create filtering handler if filtering is needed @@ -77,22 +79,39 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR filterContext.isCheckpointingDuringRecoveryEnabled(), filteringHandler, filterContext.getMemorySegmentSize())) { - read( - stateHandler, - groupByDelegate( - streamSubtaskStates(), - ChannelStateHelper::extractUnmergedInputHandles)); - read( - stateHandler, - groupByDelegate( - streamSubtaskStates(), - OperatorSubtaskState::getUpstreamOutputBufferState)); + boolean readAny = + read( + stateHandler, + groupByDelegate( + streamSubtaskStates(), + ChannelStateHelper::extractUnmergedInputHandles)); + readAny |= + read( + stateHandler, + groupByDelegate( + streamSubtaskStates(), + OperatorSubtaskState::getUpstreamOutputBufferState)); if (filteringHandler != null) { checkState( !filteringHandler.hasPartialData(), "Not all data has been fully consumed during filtering"); } + // A recovered-state container is produced whenever the checkpointing-during-recovery + // path recovered any state, regardless of whether filtering was needed: on this path + // conversion must hand the recovered buffers to the physical channels in recovery mode + // (needsRecovery = state.isPresent()), so the signal must reflect "any recovered data + // was pushed", not "a filter ran". The no-checkpointing path pushes recovered buffers + // directly and produces nothing here, matching the caller's + // checkState(readInputData(...).isEmpty()). + // + // FLINK-38544 transitional in-memory backend: recovered buffers already live in the + // physical channels' queues, so the returned container is an empty placeholder that + // only signals "there is state to recover". The spilling backend returns a real, + // file-backed container here. + return filterContext.isCheckpointingDuringRecoveryEnabled() && readAny + ? Optional.of(new FetchedChannelState()) + : Optional.empty(); } } @@ -112,15 +131,19 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR } } - private <Info, Context, Handle extends AbstractChannelStateHandle<Info>> void read( + /** Returns {@code true} if any channel state handle was read. */ + private <Info, Context, Handle extends AbstractChannelStateHandle<Info>> boolean read( RecoveredChannelStateHandler<Info, Context> stateHandler, Map<StreamStateHandle, List<Handle>> streamStateHandleListMap) throws IOException, InterruptedException { + boolean readAny = false; for (Map.Entry<StreamStateHandle, List<Handle>> delegateAndHandles : streamStateHandleListMap.entrySet()) { readSequentially( delegateAndHandles.getKey(), delegateAndHandles.getValue(), stateHandler); + readAny = true; } + return readAny; } private <Info, Context, Handle extends AbstractChannelStateHandle<Info>> void readSequentially( 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 04a5b65c2f1..31922213406 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 @@ -124,13 +124,16 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan /** * 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). + * channel in recovery state and synchronously hands every queued recovered data buffer over + * through the push interface. The legacy {@link EndOfInputChannelStateEvent} in the queue is + * dropped in translation; the {@link EndOfFetchedChannelStateEvent} sentinel takes its place + * but is deliberately NOT appended here: the StreamTask recovery chain appends it via {@link + * RecoverableInputChannel#finishRecoveredBufferDelivery()} on the channel IO executor after + * partitions have been requested. That call waits for upstream readiness, which (a) cannot + * happen on the mailbox thread that is still converting channels (partitions are requested only + * after conversion) and (b) must happen before the sentinel becomes consumable -- otherwise the + * consume path could flip the channel out of recovery and poll it before its upstream + * connection exists. */ private InputChannel toInputChannelInRecovery() throws IOException { final Buffer[] remainingBuffers; @@ -153,8 +156,6 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan recoverableChannel.onRecoveredStateBuffer(buffer); } } - recoverableChannel.onRecoveredStateBuffer( - EventSerializer.toBuffer(EndOfFetchedChannelStateEvent.INSTANCE, false)); inputChannel.checkpointStopped(lastStoppedCheckpointId); return inputChannel; } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/InputProcessorUtil.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/InputProcessorUtil.java index 6d8a0268dad..39f85fe08b2 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/InputProcessorUtil.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/InputProcessorUtil.java @@ -22,6 +22,7 @@ import org.apache.flink.api.common.operators.MailboxExecutor; import org.apache.flink.configuration.CheckpointingOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.core.execution.CheckpointingMode; +import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger; import org.apache.flink.runtime.io.network.partition.consumer.CheckpointableInput; import org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate; import org.apache.flink.runtime.io.network.partition.consumer.InputGate; @@ -88,6 +89,30 @@ public class InputProcessorUtil { List<StreamTaskSourceInput<?>> sourceInputs, MailboxExecutor mailboxExecutor, TimerService timerService) { + return createCheckpointBarrierHandler( + toNotifyOnCheckpoint, + jobConf, + config, + checkpointCoordinator, + taskName, + inputGates, + sourceInputs, + mailboxExecutor, + timerService, + RecoveryCheckpointTrigger.NO_OP); + } + + public static CheckpointBarrierHandler createCheckpointBarrierHandler( + CheckpointableTask toNotifyOnCheckpoint, + Configuration jobConf, + StreamConfig config, + SubtaskCheckpointCoordinator checkpointCoordinator, + String taskName, + List<IndexedInputGate>[] inputGates, + List<StreamTaskSourceInput<?>> sourceInputs, + MailboxExecutor mailboxExecutor, + TimerService timerService, + RecoveryCheckpointTrigger recoveryCheckpointTrigger) { CheckpointableInput[] inputs = Stream.<CheckpointableInput>concat( @@ -115,7 +140,8 @@ public class InputProcessorUtil { timerService, inputs, clock, - numberOfChannels); + numberOfChannels, + recoveryCheckpointTrigger); case AT_LEAST_ONCE: if (CheckpointingOptions.isUnalignedCheckpointEnabled(jobConf)) { throw new IllegalStateException( @@ -148,7 +174,8 @@ public class InputProcessorUtil { TimerService timerService, CheckpointableInput[] inputs, Clock clock, - int numberOfChannels) { + int numberOfChannels, + RecoveryCheckpointTrigger recoveryCheckpointTrigger) { boolean enableCheckpointAfterTasksFinished = config.getConfiguration() .get(CheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH); @@ -161,6 +188,7 @@ public class InputProcessorUtil { numberOfChannels, BarrierAlignmentUtil.createRegisterTimerCallback(mailboxExecutor, timerService), enableCheckpointAfterTasksFinished, + recoveryCheckpointTrigger, inputs); } else { return SingleCheckpointBarrierHandler.aligned( diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java index c1bd9ad6c85..8594fba540f 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java @@ -23,6 +23,7 @@ import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.runtime.checkpoint.CheckpointException; import org.apache.flink.runtime.checkpoint.CheckpointFailureReason; import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; +import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger; import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker; import org.apache.flink.runtime.io.network.api.CheckpointBarrier; import org.apache.flink.runtime.io.network.partition.consumer.CheckpointableInput; @@ -119,6 +120,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler { "Strictly unaligned checkpoints should never register any callbacks"); }, enableCheckpointsAfterTasksFinish, + RecoveryCheckpointTrigger.NO_OP, inputs); } @@ -130,6 +132,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler { int numOpenChannels, DelayableTimer registerTimer, boolean enableCheckpointAfterTasksFinished, + RecoveryCheckpointTrigger recoveryCheckpointTrigger, CheckpointableInput... inputs) { return new SingleCheckpointBarrierHandler( taskName, @@ -137,7 +140,8 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler { checkpointCoordinator, clock, numOpenChannels, - new AlternatingWaitingForFirstBarrierUnaligned(false, new ChannelState(inputs)), + new AlternatingWaitingForFirstBarrierUnaligned( + false, new ChannelState(inputs, recoveryCheckpointTrigger)), false, registerTimer, inputs, @@ -173,6 +177,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler { int numOpenChannels, DelayableTimer registerTimer, boolean enableCheckpointAfterTasksFinished, + RecoveryCheckpointTrigger recoveryCheckpointTrigger, CheckpointableInput... inputs) { return new SingleCheckpointBarrierHandler( taskName, @@ -180,7 +185,8 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler { checkpointCoordinator, clock, numOpenChannels, - new AlternatingWaitingForFirstBarrier(new ChannelState(inputs)), + new AlternatingWaitingForFirstBarrier( + new ChannelState(inputs, recoveryCheckpointTrigger)), true, registerTimer, inputs, diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java index 6e45024e1ca..e27876eccd7 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java @@ -149,7 +149,8 @@ public class MultipleInputStreamTask<OUT> inputGates, operatorChain.getSourceTaskInputs(), mainMailboxExecutor, - timerService); + timerService, + getRecoveryCheckpointTrigger()); CheckpointedInputGate[] checkpointedInputGates = InputProcessorUtil.createCheckpointedMultipleInputGate( diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java index 009f48b082f..eb772462f1f 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java @@ -174,7 +174,8 @@ public class OneInputStreamTask<IN, OUT> extends StreamTask<OUT, OneInputStreamO new List[] {Arrays.asList(inputGates)}, Collections.emptyList(), mainMailboxExecutor, - systemTimerService); + systemTimerService, + getRecoveryCheckpointTrigger()); CheckpointedInputGate[] checkpointedInputGates = InputProcessorUtil.createCheckpointedMultipleInputGate( diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java index 6d776afb1cd..d9d6738caf8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java @@ -43,7 +43,10 @@ import org.apache.flink.runtime.checkpoint.SavepointType; import org.apache.flink.runtime.checkpoint.SnapshotType; import org.apache.flink.runtime.checkpoint.SubTaskInitializationMetricsBuilder; import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter; +import org.apache.flink.runtime.checkpoint.channel.FetchedChannelState; +import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateDrainer; import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; +import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger; import org.apache.flink.runtime.checkpoint.channel.SequentialChannelStateReader; import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager; import org.apache.flink.runtime.execution.CancelTaskException; @@ -309,8 +312,13 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> /** TODO it might be replaced by the global IO executor on TaskManager level future. */ private final ExecutorService channelIOExecutor; + private RecoveryCheckpointTrigger recoveryCheckpointTrigger = + RecoveryCheckpointTrigger.NOT_READY; + /** - * Completes when channel recovery has finished; set by {@link #restoreStateAndGates}. Used to + * Completes when the restore phase may end; set by {@link #restoreStateAndGates}. With + * checkpointing during recovery this is the point where the recovery checkpoint trigger is + * installed, while draining and consuming the fetched state continue in the background. Used to * defer the lifecycle finish of a task that reaches {@code END_OF_INPUT} during recovery (e.g. * a bounded operator whose input is already fully available) so that it does not suspend the * restore mailbox loop before recovery completes. Accessed only on the mailbox/task thread. @@ -930,26 +938,70 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> private CompletableFuture<Void> recoverChannelsWithCheckpointing( SequentialChannelStateReader reader, IndexedInputGate[] inputGates) { if (inputGates.length == 0) { + recoveryCheckpointTrigger = RecoveryCheckpointTrigger.NO_OP; return FutureUtils.completedVoidFuture(); } - // FLINK-38544 transitional: removed when the spilling backend lands. In-memory backend: - // readInputChannelState first filters the recovered state completely (the filter output - // accumulates in the recovered channels' in-memory queues), then conversion pushes it into - // the physical channels through the push interface (requestPartitions(true)). The final - // form instead fetches the filter output to disk and drains it incrementally from here. - return CompletableFuture.runAsync( - () -> readInputChannelState(reader, inputGates), channelIOExecutor) - .thenCompose(ign -> requestPartitions(inputGates, true)) + return setRecoveryCheckpointTrigger(RecoveryCheckpointTrigger.NOT_READY) + .thenApplyAsync(ign -> fetchChannelState(reader, inputGates), channelIOExecutor) + .thenCompose( + state -> + requestPartitions(inputGates, state.isPresent()) + .thenApply(channels -> buildDrainer(state, channels))) .thenCompose( - ign -> - completeAll( - Arrays.stream(inputGates) - .map(InputGate::getStateConsumedFuture) - .collect(Collectors.toList()))); + drainerOpt -> { + if (drainerOpt.isEmpty()) { + return setRecoveryCheckpointTrigger( + RecoveryCheckpointTrigger.NO_OP); + } + FetchedChannelStateDrainer drainer = drainerOpt.get(); + CompletableFuture<Void> triggerInstalled = + setRecoveryCheckpointTrigger(drainer); + triggerInstalled + .thenRunAsync(() -> drain(drainer), channelIOExecutor) + .thenCompose( + ign -> + completeAll( + Arrays.stream(inputGates) + .map( + InputGate + ::getStateConsumedFuture) + .collect(Collectors.toList()))) + .thenCompose( + ign -> + setRecoveryCheckpointTrigger( + RecoveryCheckpointTrigger.NO_OP)) + .exceptionally( + t -> { + asyncExceptionHandler.handleAsyncException( + "Unable to finalize recovered channel state consumption", + t); + return null; + }); + return triggerInstalled; + }); + } + + @SuppressWarnings("OptionalUsedAsFieldOrParameterType") // intentional: simplify call-site + private Optional<FetchedChannelStateDrainer> buildDrainer( + Optional<FetchedChannelState> state, List<RecoverableInputChannel> channels) { + return state.map(s -> new FetchedChannelStateDrainer(s, channels)); + } + + private CompletableFuture<Void> setRecoveryCheckpointTrigger( + RecoveryCheckpointTrigger trigger) { + CompletableFuture<Void> future = new CompletableFuture<>(); + mainMailboxExecutor.execute( + () -> { + recoveryCheckpointTrigger = trigger; + future.complete(null); + }, + "update recoveryCheckpointTrigger to " + trigger); + return future; } private CompletableFuture<Void> recoverChannelsWithoutCheckpointing( SequentialChannelStateReader reader, IndexedInputGate[] inputGates) { + recoveryCheckpointTrigger = RecoveryCheckpointTrigger.NO_OP; // Feed recovered channel state on the IO thread. This is intentionally NOT part of the // completion gate below: a gate's stateConsumedFuture only completes once the consumer (the // default action, running in the restore mailbox loop) has drained the end-of-state @@ -982,7 +1034,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> private void readInputChannelState( SequentialChannelStateReader reader, IndexedInputGate[] inputGates) { try { - reader.readInputData(inputGates, createRecordFilterContext()); + checkState(reader.readInputData(inputGates, createRecordFilterContext()).isEmpty()); for (IndexedInputGate gate : inputGates) { gate.finishReadRecoveredState(); @@ -993,6 +1045,31 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> } } + private Optional<FetchedChannelState> fetchChannelState( + SequentialChannelStateReader reader, IndexedInputGate[] inputGates) { + try { + return reader.readInputData(inputGates, createRecordFilterContext()); + } catch (Throwable t) { + asyncExceptionHandler.handleAsyncException( + "Unable to set up recovered channel state", t); + return Optional.empty(); + } + } + + private void drain(FetchedChannelStateDrainer drainer) { + try (FetchedChannelStateDrainer ignored = drainer) { + try { + drainer.drain(); + } catch (Throwable t) { + asyncExceptionHandler.handleAsyncException( + "Unable to drain recovered channel state", t); + } + } catch (Throwable closeError) { + asyncExceptionHandler.handleAsyncException( + "Unable to close FetchedChannelStateDrainer after drain", closeError); + } + } + private CompletableFuture<List<RecoverableInputChannel>> requestPartitions( IndexedInputGate[] inputGates, boolean needsRecovery) { CompletableFuture<List<RecoverableInputChannel>> future = new CompletableFuture<>(); @@ -1026,6 +1103,13 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> return channels; } + public RecoveryCheckpointTrigger getRecoveryCheckpointTrigger() { + return cpId -> { + checkState(mailboxProcessor.isMailboxThread()); + recoveryCheckpointTrigger.snapshotAndInsertBarriers(cpId); + }; + } + private void ensureNotCanceled() { if (canceled) { throw new CancelTaskException(); @@ -2098,6 +2182,17 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> // For source tasks, this will be 0. For tasks with network inputs, each physical gate // must have a corresponding config entry. int numGates = getEnvironment().getAllInputGates().length; + + if (numGates > 0 && inEdges.isEmpty()) { + // The task has input gates but the StreamConfig carries no physical edges. This happens + // for dynamically connected inputs -- e.g. reading a cached intermediate result under + // the AdaptiveBatchScheduler -- where the physical connection (and thus the partitioner + // needed to build per-gate filter configs) is determined by the scheduler at runtime. + // Such jobs are batch and have no unaligned-checkpoint channel state to filter, so + // disable record filtering rather than failing the precondition below. + return RecordFilterContext.disabled(); + } + RecordFilterContext.InputFilterConfig[] inputConfigs = new RecordFilterContext.InputFilterConfig[numGates]; diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java index f933d5069d1..5ea0f544285 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/TwoInputStreamTask.java @@ -72,7 +72,8 @@ public class TwoInputStreamTask<IN1, IN2, OUT> extends AbstractTwoInputStreamTas new List[] {inputGates1, inputGates2}, Collections.emptyList(), mainMailboxExecutor, - systemTimerService); + systemTimerService, + getRecoveryCheckpointTrigger()); CheckpointedInputGate[] checkpointedInputGates = InputProcessorUtil.createCheckpointedMultipleInputGate( 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 87a6c0466de..15f1b8dc925 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,7 +22,6 @@ 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; @@ -110,17 +109,15 @@ class RecoveredInputChannelTest { 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); + // The queued data buffer is handed over through the push interface and the legacy + // EndOfInputChannelStateEvent is dropped in translation. The EndOfFetchedChannelStateEvent + // sentinel is deliberately NOT appended here: the StreamTask recovery chain appends it via + // finishRecoveredBufferDelivery() on the channel IO executor after partitions have been + // requested (the sentinel must not become consumable before upstream readiness). + assertThat(converted.getRecoveredBuffersSpy()).hasSize(1); 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); } @Test diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/TestBarrierHandlerFactory.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/TestBarrierHandlerFactory.java index 1b345ebe0d2..dca71e35f6b 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/TestBarrierHandlerFactory.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/TestBarrierHandlerFactory.java @@ -20,6 +20,7 @@ package org.apache.flink.streaming.runtime.io.checkpointing; import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter; import org.apache.flink.runtime.checkpoint.channel.RecordingChannelStateWriter; +import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger; import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate; import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; import org.apache.flink.streaming.runtime.tasks.TestSubtaskCheckpointCoordinator; @@ -72,6 +73,7 @@ public class TestBarrierHandlerFactory { inputGate.getNumberOfInputChannels(), actionRegistration, enableCheckpointsAfterTasksFinish, + RecoveryCheckpointTrigger.NO_OP, inputGate); } }
