rkhachatryan commented on code in PR #28660:
URL: https://github.com/apache/flink/pull/28660#discussion_r3605788192
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java:
##########
@@ -98,4 +111,26 @@ public ChannelState emptyState() {
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();
+ try {
+ recoveryCheckpointTrigger.snapshotAndInsertBarriers(cpId);
+
+ for (CheckpointableInput input : inputs) {
+ input.checkpointStarted(barrier);
+ }
+ } catch (Throwable t) {
+ if (t instanceof CheckpointException) {
+ throw (CheckpointException) t;
+ }
+ ExceptionUtils.rethrowIOException(t);
+ }
Review Comment:
Why do we need to catch and rethrow Throwable?
I think the whole try/catch here is unnecessary here: the body already
declares CheckpointException, IOException; and the callers are catching
Exception
##########
flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCollectingBarriersDispatchHookTest.java:
##########
@@ -0,0 +1,124 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Source-level invariants for the {@code alignedCheckpointTimeout} → UC
switch in {@link
+ * AlternatingCollectingBarriers}. Note this is NOT a {@code barrierReceived}
hook — the parent
+ * {@code AbstractAlternatingAlignedBarrierHandlerState.barrierReceived} is
{@code final}; the only
+ * place to switch from aligned to UC is {@code alignedCheckpointTimeout}.
+ */
+class AlternatingCollectingBarriersDispatchHookTest {
+
+ @Test
+ void testDispatcherIsCalledBeforeTriggerGlobalCheckpoint() throws
Exception {
+ String source = readSource();
+
+ int initIdx =
source.indexOf("controller.initInputsCheckpoint(unalignedBarrier)");
+ int dispatchIdx =
source.indexOf("onCheckpointStartedForAllInputs(unalignedBarrier)");
+ int triggerIdx =
source.indexOf("controller.triggerGlobalCheckpoint(unalignedBarrier)");
+
+ assertThat(initIdx).as("initInputsCheckpoint call
exists").isNotNegative();
+ assertThat(dispatchIdx).as("dispatcher call exists").isNotNegative();
+ assertThat(triggerIdx).as("triggerGlobalCheckpoint call
exists").isNotNegative();
+
+ assertThat(initIdx)
+ .as(
+ "initInputsCheckpoint precedes dispatcher (cpId result
registered before "
+ + "Step 3)")
+ .isLessThan(dispatchIdx);
+ assertThat(dispatchIdx)
+ .as("dispatcher precedes triggerGlobalCheckpoint (master
ordering)")
+ .isLessThan(triggerIdx);
+ }
+
+ @Test
+ void testHookIsInAlignedCheckpointTimeoutNotBarrierReceived() throws
Exception {
+ String source = readSource();
+
+ // Hard guard:
AbstractAlternatingAlignedBarrierHandlerState.barrierReceived is final, so
+ // AlternatingCollectingBarriers cannot override it. The UC switch
lives in
+ // alignedCheckpointTimeout only.
+ assertThat(source)
+ .as("alignedCheckpointTimeout is the UC switch entry on this
state class")
+ .contains("public BarrierHandlerState
alignedCheckpointTimeout(");
+ assertThat(source)
+ .as("No barrierReceived override should exist on
AlternatingCollectingBarriers")
+ .doesNotContain("public BarrierHandlerState barrierReceived(");
+
+ int timeoutIdx = source.indexOf("alignedCheckpointTimeout(");
+ int dispatchIdx = source.indexOf("onCheckpointStartedForAllInputs(",
timeoutIdx);
+ assertThat(dispatchIdx)
+ .as("dispatcher call is inside alignedCheckpointTimeout body")
+ .isNotNegative();
+ }
+
+ @Test
+ void testNoLegacyPerInputCheckpointStartedLoopInTimeout() throws Exception
{
+ String source = readSource();
+
+ int methodStart = source.indexOf("public BarrierHandlerState
alignedCheckpointTimeout(");
+ assertThat(methodStart).isNotNegative();
+ int methodEnd = findMethodEnd(source, methodStart);
+ String body = source.substring(methodStart, methodEnd);
+
+ assertThat(body)
+ .as(
+ "Pre-trigger per-input checkpointStarted loop should
be removed; dispatcher "
+ + "now owns the fan-out")
+ .doesNotContain("input.checkpointStarted(unalignedBarrier)");
+ }
+
+ private static String readSource() throws Exception {
+ Path candidate =
+ Paths.get(
+
"src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCollectingBarriers.java");
+ if (!Files.exists(candidate)) {
+ candidate =
+ Paths.get(
+
"flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/AlternatingCollectingBarriers.java");
+ }
+ return new String(Files.readAllBytes(candidate));
+ }
Review Comment:
I don't think this testing methodology is suitable for Apache Flink.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
