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 77ecb4dddc06fa67ea0057d6031fdb2f8e6ffae6 Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 01:23:25 2026 +0200 [FLINK-39523][checkpoint] ChannelState: dispatch checkpoint start through the recovery trigger ChannelState gains a RecoveryCheckpointTrigger (the legacy 1-arg ctor defaults to NO_OP); new onCheckpointStartedForAllInputs(CheckpointBarrier): (1) trigger.snapshotAndInsertBarriers(cpId); (2) for (input : inputs) input.checkpointStarted(barrier). CheckpointException is rethrown as-is (routes to checkpoint abort, not task failure); other IOException via rethrowIOException. (Step 3 -- spilled-slice replay through the channel-state writer -- is added when the spilling backend lands.) AlternatingCollectingBarriers and AlternatingWaitingForFirstBarrierUnaligned: replace the inline per-input checkpointStarted loop with state.onCheckpointStartedForAllInputs(...) (behaviorally inert with NO_OP). Tests: ChannelStateTest (2-step scope), AlternatingCollectingBarriersDispatchHookTest, AlternatingWaitingForFirstBarrierUnalignedDispatchHookTest. --- .../AlternatingCollectingBarriers.java | 5 +- ...AlternatingWaitingForFirstBarrierUnaligned.java | 4 +- .../runtime/io/checkpointing/ChannelState.java | 27 ++++ .../runtime/io/checkpointing/ChannelStateTest.java | 147 +++++++++++++++++++++ 4 files changed, 176 insertions(+), 7 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCollectingBarriers.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCollectingBarriers.java index 8ca37055bc3..c918f9db0ee 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCollectingBarriers.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCollectingBarriers.java @@ -21,7 +21,6 @@ package org.apache.flink.streaming.runtime.io.checkpointing; import org.apache.flink.runtime.checkpoint.CheckpointException; import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; import org.apache.flink.runtime.io.network.api.CheckpointBarrier; -import org.apache.flink.runtime.io.network.partition.consumer.CheckpointableInput; import java.io.IOException; @@ -44,9 +43,7 @@ final class AlternatingCollectingBarriers extends AbstractAlternatingAlignedBarr state.prioritizeAllAnnouncements(); CheckpointBarrier unalignedBarrier = checkpointBarrier.asUnaligned(); controller.initInputsCheckpoint(unalignedBarrier); - for (CheckpointableInput input : state.getInputs()) { - input.checkpointStarted(unalignedBarrier); - } + state.onCheckpointStartedForAllInputs(unalignedBarrier); controller.triggerGlobalCheckpoint(unalignedBarrier); return new AlternatingCollectingBarriersUnaligned(true, state); } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingWaitingForFirstBarrierUnaligned.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingWaitingForFirstBarrierUnaligned.java index af04f4f8107..1e12b4757e8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingWaitingForFirstBarrierUnaligned.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingWaitingForFirstBarrierUnaligned.java @@ -72,9 +72,7 @@ final class AlternatingWaitingForFirstBarrierUnaligned implements BarrierHandler CheckpointBarrier unalignedBarrier = checkpointBarrier.asUnaligned(); controller.initInputsCheckpoint(unalignedBarrier); - for (CheckpointableInput input : channelState.getInputs()) { - input.checkpointStarted(unalignedBarrier); - } + channelState.onCheckpointStartedForAllInputs(unalignedBarrier); controller.triggerGlobalCheckpoint(unalignedBarrier); if (controller.allBarriersReceived()) { for (CheckpointableInput input : channelState.getInputs()) { diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java index 8f6bb211d2b..503e1532724 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java @@ -18,7 +18,9 @@ package org.apache.flink.streaming.runtime.io.checkpointing; +import org.apache.flink.runtime.checkpoint.CheckpointException; import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; +import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger; import org.apache.flink.runtime.io.network.api.CheckpointBarrier; import org.apache.flink.runtime.io.network.partition.consumer.CheckpointableInput; @@ -28,6 +30,7 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import static org.apache.flink.util.Preconditions.checkNotNull; import static org.apache.flink.util.Preconditions.checkState; /** @@ -35,6 +38,7 @@ import static org.apache.flink.util.Preconditions.checkState; * and {@link AbstractAlternatingAlignedBarrierHandlerState}. */ final class ChannelState { + private final Map<InputChannelInfo, Integer> sequenceNumberInAnnouncedChannels = new HashMap<>(); @@ -47,8 +51,16 @@ final class ChannelState { private final CheckpointableInput[] inputs; + private final RecoveryCheckpointTrigger recoveryCheckpointTrigger; + public ChannelState(CheckpointableInput[] inputs) { + this(inputs, RecoveryCheckpointTrigger.NO_OP); + } + + public ChannelState( + CheckpointableInput[] inputs, RecoveryCheckpointTrigger recoveryCheckpointTrigger) { this.inputs = inputs; + this.recoveryCheckpointTrigger = checkNotNull(recoveryCheckpointTrigger); } public void blockChannel(InputChannelInfo channelInfo) { @@ -98,4 +110,19 @@ final class ChannelState { sequenceNumberInAnnouncedChannels.clear(); return this; } + + /** + * Dispatches checkpoint start: inserts recovery-checkpoint barriers into in-recovery channels + * through the trigger, then notifies every input. (FLINK-38544 transitional: the spilling + * backend adds a third step handing the trigger's snapshot reader to the channel-state writer.) + */ + public void onCheckpointStartedForAllInputs(CheckpointBarrier barrier) + throws CheckpointException, IOException { + long cpId = barrier.getId(); + recoveryCheckpointTrigger.snapshotAndInsertBarriers(cpId); + + for (CheckpointableInput input : inputs) { + input.checkpointStarted(barrier); + } + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelStateTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelStateTest.java new file mode 100644 index 00000000000..b0e21aef390 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelStateTest.java @@ -0,0 +1,147 @@ +/* + * 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.streaming.runtime.io.checkpointing; + +import org.apache.flink.runtime.checkpoint.CheckpointException; +import org.apache.flink.runtime.checkpoint.CheckpointOptions; +import org.apache.flink.runtime.checkpoint.CheckpointType; +import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; +import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger; +import org.apache.flink.runtime.io.network.api.CheckpointBarrier; +import org.apache.flink.runtime.io.network.partition.consumer.CheckpointableInput; +import org.apache.flink.runtime.state.CheckpointStorageLocationReference; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Verifies the {@link ChannelState#onCheckpointStartedForAllInputs} dispatcher: call ordering and + * feature-off no-op routing through the {@link RecoveryCheckpointTrigger#NO_OP} singleton. + * + * <p>FLINK-38544 transitional: this covers the 2-step dispatch (trigger, then per-input + * notification); the spilling backend adds a third step handing the trigger's snapshot reader to + * the channel-state writer and completes this test to cover all three steps. + */ +class ChannelStateTest { + + private static final long CHECKPOINT_ID = 7L; + + @Test + void testStepOrderingFeatureOn() throws Exception { + List<String> trace = new ArrayList<>(); + RecordingTrigger trigger = new RecordingTrigger(trace); + CheckpointableInput input1 = new RecordingInput(trace, "in1"); + CheckpointableInput input2 = new RecordingInput(trace, "in2"); + + ChannelState state = new ChannelState(new CheckpointableInput[] {input1, input2}, trigger); + + CheckpointBarrier barrier = newUnalignedBarrier(); + state.onCheckpointStartedForAllInputs(barrier); + + assertThat(trace) + .containsExactly( + "trigger.snapshotAndInsertBarriers:" + CHECKPOINT_ID, + "in1.checkpointStarted:" + CHECKPOINT_ID, + "in2.checkpointStarted:" + CHECKPOINT_ID); + } + + @Test + void testStepOrderingFeatureOff() throws Exception { + List<String> trace = new ArrayList<>(); + CheckpointableInput input = new RecordingInput(trace, "in1"); + + ChannelState state = + new ChannelState( + new CheckpointableInput[] {input}, RecoveryCheckpointTrigger.NO_OP); + + state.onCheckpointStartedForAllInputs(newUnalignedBarrier()); + + assertThat(trace).containsExactly("in1.checkpointStarted:" + CHECKPOINT_ID); + } + + private static CheckpointBarrier newUnalignedBarrier() { + return new CheckpointBarrier( + CHECKPOINT_ID, + 1000L, + CheckpointOptions.unaligned( + CheckpointType.CHECKPOINT, + CheckpointStorageLocationReference.getDefault())); + } + + private static final class RecordingTrigger implements RecoveryCheckpointTrigger { + private final List<String> trace; + + RecordingTrigger(List<String> trace) { + this.trace = trace; + } + + @Override + public void snapshotAndInsertBarriers(long checkpointId) { + trace.add("trigger.snapshotAndInsertBarriers:" + checkpointId); + } + } + + private static final class RecordingInput implements CheckpointableInput { + + private final List<String> trace; + private final String name; + + RecordingInput(List<String> trace, String name) { + this.trace = trace; + this.name = name; + } + + @Override + public void blockConsumption(InputChannelInfo channelInfo) {} + + @Override + public void resumeConsumption(InputChannelInfo channelInfo) {} + + @Override + public List<InputChannelInfo> getChannelInfos() { + return Collections.emptyList(); + } + + @Override + public int getNumberOfInputChannels() { + return 0; + } + + @Override + public void checkpointStarted(CheckpointBarrier barrier) throws CheckpointException { + trace.add(name + ".checkpointStarted:" + barrier.getId()); + } + + @Override + public void checkpointStopped(long cancelledCheckpointId) {} + + @Override + public int getInputGateIndex() { + return 0; + } + + @Override + public void convertToPriorityEvent(int channelIndex, int sequenceNumber) {} + } +}
