(flink) 02/02: [FLINK-40081][test] Add CdrRecoveryRaceITCase for the non-source recovery race

[fa...@apache.org]

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 2f3c205e9266cb30240eb7f4fdab15cad629a70f
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Mon Jul 6 12:35:02 2026 +0200

    [FLINK-40081][test] Add CdrRecoveryRaceITCase for the non-source recovery race
    
    Adds a bounded BATCH job (blocking exchange -> non-source operator) that
    exercises the checkpointing-during-recovery restore path for an operator
    whose input is fully available, i.e. the scenario that tripped "Mailbox
    loop interrupted before recovery was finished" on TPC-DS and is fixed by
    the defer-task-finish commit of this series.
    
    INCOMPLETE -- does not reliably guard the regression. The race window on
    a fresh deployment with no recovered channel state is tiny, so this test
    does NOT deterministically reproduce the bug on its own. During
    development it reproduced only with a temporary delay injected into the
    recovery chain to widen the window; that delay is not product code and
    is not included here. As committed it is a fast BATCH + CDR smoke test
    of the no-recovered-state recovery path, not a guaranteed reproducer.
    
    Follow-up (see TODO in the test): make it deterministic via a test-only
    hook to widen the recovery window, or an unaligned-checkpoint restore
    that carries in-flight channel state, before treating it as a real
    regression test.
---
 .../test/checkpointing/CdrRecoveryRaceITCase.java  | 89 ++++++++++++++++++++++
 1 file changed, 89 insertions(+)

diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CdrRecoveryRaceITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CdrRecoveryRaceITCase.java
new file mode 100644
index 00000000000..feeeeb5544d
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CdrRecoveryRaceITCase.java
@@ -0,0 +1,89 @@
+/*
+ * 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.test.checkpointing;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.CheckpointingMode;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * Exercises the checkpointing-during-recovery restore-loop path for NON-source operators: a bounded
+ * BATCH operator (behind a blocking exchange) whose input is already fully available drains it to
+ * {@code END_OF_INPUT} during the recovery window. Without the fix in {@code
+ * StreamTask.recoverChannels} this suspends the restore mailbox loop before recovery completes,
+ * tripping "Mailbox loop interrupted before recovery was finished" in {@code
+ * StreamTask.restoreInternal}. CDR is forced on by the surrounding test setup.
+ *
+ * <p>INCOMPLETE / NOT A RELIABLE REGRESSION GUARD. The race window on a fresh deployment with no
+ * recovered channel state is very small, so this test does NOT deterministically reproduce the bug
+ * on its own: during development it reproduced reliably only with a temporary artificial delay
+ * inserted into the recovery chain (after partitions are requested, before completion) to widen the
+ * window -- that delay is intentionally NOT part of the product code. As committed this is a fast
+ * BATCH + CDR smoke test of the no-recovered-state recovery path, not a guaranteed reproducer; a
+ * future iteration should make it deterministic (e.g. a test-only hook to widen the window, or an
+ * unaligned-checkpoint restore that carries in-flight channel state) before relying on it.
+ */
+// TODO FLINK-38544: make this deterministically reproduce the non-source recovery race.
+class CdrRecoveryRaceITCase {
+
+    private static final int PARALLELISM = 8;
+    private static final int ITERATIONS = 30;
+
+    @RegisterExtension
+    private static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+            new MiniClusterExtension(
+                    new MiniClusterResourceConfiguration.Builder()
+                            .setConfiguration(new Configuration())
+                            .setNumberTaskManagers(2)
+                            .setNumberSlotsPerTaskManager(PARALLELISM)
+                            .build());
+
+    @Test
+    void testBoundedOperatorFinishesDuringRecovery() throws Exception {
+        for (int i = 0; i < ITERATIONS; i++) {
+            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+            env.setParallelism(PARALLELISM);
+            // BATCH mode: blocking exchanges make a downstream operator's entire input available
+            // before it deploys, so during its restore loop it drains straight to END_OF_INPUT --
+            // the condition that trips the non-source recovery race (mirrors TPC-DS).
+            env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+            env.enableCheckpointing(50, CheckpointingMode.EXACTLY_ONCE);
+
+            DataStream<Long> source = env.fromSequence(1, 64);
+            source.rebalance()
+                    .map((MapFunction<Long, Long>) value -> value + 1)
+                    .name("non-source-map")
+                    .rebalance()
+                    .map((MapFunction<Long, Long>) value -> value * 2)
+                    .name("non-source-map-2")
+                    .sinkTo(new DiscardingSink<>());
+
+            env.execute("cdr-recovery-race-" + i);
+        }
+    }
+}