This is an automated email from the ASF dual-hosted git repository.
rkhachatryan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new d2652156875 [FLINK-40200][Checkpoint] Checkpoint sync phase timeout
d2652156875 is described below
commit d2652156875b26eeb93ec08ff5c9204bcba78112
Author: Efrat Levitan <[email protected]>
AuthorDate: Thu Jul 16 14:25:47 2026 +0300
[FLINK-40200][Checkpoint] Checkpoint sync phase timeout
For TaskManagers locking up in rocksDB native calls during checkpoint sync
phase,
the best mitigation would be to cancel the task and restart.
A lenient checkpoint timeout X
{execution.checkpointing.tolerable-failed-checkpoints}
means it can take days for JM to restart the job out of the deadlock.
Setting checkpoint sync timeout, after which a task failure is triggered,
will address thread bloackages faster and without manual intervention.
---
.../generated/checkpointing_configuration.html | 6 +
.../flink/configuration/CheckpointingOptions.java | 13 +++
.../org/apache/flink/runtime/taskmanager/Task.java | 9 +-
.../api/environment/CheckpointConfig.java | 14 +++
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 72 ++++++++++++
.../tasks/SubtaskCheckpointCoordinatorTest.java | 126 +++++++++++++++++++++
.../streaming/util/TestStreamEnvironment.java | 5 +
.../CheckpointSyncPhaseTimeoutITCase.java | 93 +++++++++++++++
8 files changed, 335 insertions(+), 3 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/checkpointing_configuration.html
b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
index 2fcbc5fc1ed..c256e4a66e6 100644
--- a/docs/layouts/shortcodes/generated/checkpointing_configuration.html
+++ b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
@@ -152,6 +152,12 @@
<td>String</td>
<td>The checkpoint storage implementation to be used to checkpoint
state.<br />The implementation can be specified either via their shortcut
name, or via the class name of a <code
class="highlighter-rouge">CheckpointStorageFactory</code>. If a factory is
specified it is instantiated via its zero argument constructor and its <code
class="highlighter-rouge">CheckpointStorageFactory#createFromConfig(ReadableConfig,
ClassLoader)</code> method is called.<br />Recognized shortcut [...]
</tr>
+ <tr>
+ <td><h5>execution.checkpointing.sync-phase-timeout</h5></td>
+ <td style="word-wrap: break-word;">0 ms</td>
+ <td>Duration</td>
+ <td>A timeout for the synchronous phase of a checkpoint, after
which a task cancellation is triggered. If the task thread is merely slow, the
task will restart and continue. If a true thread blockage is encountered, (e.g.
by a blocking native call), a fatal error will be thrown after the
task.cancellation.timeoutand trigger a TM restart to address the blockage.
Defaults to 0 (disabled).</td>
+ </tr>
<tr>
<td><h5>execution.checkpointing.timeout</h5></td>
<td style="word-wrap: break-word;">10 min</td>
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
index 75abd4178a0..41397521f84 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java
@@ -583,6 +583,19 @@ public class CheckpointingOptions {
+ "will timeout and
checkpoint barrier will start working as unaligned checkpoint.")
.build());
+ public static final ConfigOption<Duration>
CHECKPOINTING_SYNC_PHASE_TIMEOUT =
+ ConfigOptions.key("execution.checkpointing.sync-phase-timeout")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(0L))
+ .withDescription(
+ "A timeout for the synchronous phase of a
checkpoint, after which a task cancellation is triggered."
+ + " If the task thread is merely slow, the
task will restart and continue."
+ + " If a true thread blockage is
encountered, (e.g. by a blocking native call), "
+ + " a fatal error will be thrown after the
"
+ +
TaskManagerOptions.TASK_CANCELLATION_TIMEOUT.key()
+ + "and trigger a TM restart to address the
blockage."
+ + " Defaults to 0 (disabled).");
+
public static final ConfigOption<Boolean> FORCE_UNALIGNED =
ConfigOptions.key("execution.checkpointing.unaligned.forced")
.booleanType()
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index 249f29af1be..3ef341c1372 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -1879,20 +1879,23 @@ public class Task
}
}
- public static void logTaskThreadStackTrace(
- Thread thread, String taskName, long timeoutMs, String action) {
+ public static StringBuilder getTaskThreadStackTrace(Thread thread) {
StackTraceElement[] stack = thread.getStackTrace();
StringBuilder stackTraceStr = new StringBuilder();
for (StackTraceElement e : stack) {
stackTraceStr.append(e).append('\n');
}
+ return stackTraceStr;
+ }
+ public static void logTaskThreadStackTrace(
+ Thread thread, String taskName, long timeoutMs, String action) {
LOG.warn(
"Task '{}' did not react to cancelling signal - {}; it is
stuck for {} seconds in method:\n {}",
taskName,
action,
timeoutMs / 1000,
- stackTraceStr);
+ getTaskThreadStackTrace(thread));
}
/** Various operation of notify checkpoint. */
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
index 8d0982a772d..88d47fd245d 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java
@@ -498,6 +498,17 @@ public class CheckpointConfig implements
java.io.Serializable {
CheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT,
alignedCheckpointTimeout);
}
+ @Experimental
+ public Duration getCheckpointSyncPhaseTimeout() {
+ return
configuration.get(CheckpointingOptions.CHECKPOINTING_SYNC_PHASE_TIMEOUT);
+ }
+
+ @Experimental
+ public void setCheckpointSyncPhaseTimeout(Duration
checkpointSyncPhaseTimeout) {
+ configuration.set(
+ CheckpointingOptions.CHECKPOINTING_SYNC_PHASE_TIMEOUT,
checkpointSyncPhaseTimeout);
+ }
+
/**
* @return the number of subtasks to share the same channel state file, as
configured via {@link
* #setMaxSubtasksPerChannelStateFile(int)} or {@link
@@ -643,6 +654,9 @@ public class CheckpointConfig implements
java.io.Serializable {
configuration
.getOptional(CheckpointingOptions.PAUSE_SOURCES_UNTIL_FIRST_CHECKPOINT)
.ifPresent(this::setPauseSourcesUntilFirstCheckpoint);
+ configuration
+
.getOptional(CheckpointingOptions.CHECKPOINTING_SYNC_PHASE_TIMEOUT)
+ .ifPresent(this::setCheckpointSyncPhaseTimeout);
}
public void setPauseSourcesUntilFirstCheckpoint(boolean
pauseCheckpointsIfTasksNotRunning) {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index 811354c47f3..d4debba167c 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder;
@@ -37,6 +38,7 @@ import
org.apache.flink.runtime.state.CheckpointStorageWorkerView;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import
org.apache.flink.runtime.state.filesystem.FsMergingCheckpointStorageLocation;
import org.apache.flink.runtime.taskmanager.AsyncExceptionHandler;
+import org.apache.flink.runtime.taskmanager.AsynchronousException;
import org.apache.flink.runtime.taskmanager.Task;
import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
import
org.apache.flink.streaming.runtime.io.checkpointing.BarrierAlignmentUtil;
@@ -44,6 +46,7 @@ import
org.apache.flink.streaming.runtime.io.checkpointing.BarrierAlignmentUtil.
import
org.apache.flink.streaming.runtime.io.checkpointing.BarrierAlignmentUtil.DelayableTimer;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FatalExitExceptionHandler;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.clock.Clock;
@@ -70,7 +73,10 @@ import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -132,6 +138,8 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
@Nullable private final FileMergingSnapshotManager
fileMergingSnapshotManager;
+ private final long syncPhaseTimeoutMillis;
+
@VisibleForTesting
SubtaskCheckpointCoordinatorImpl(
CheckpointStorageWorkerView checkpointStorage,
@@ -197,6 +205,16 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
this.registerTimer = registerTimer;
this.clock = SystemClock.getInstance();
this.fileMergingSnapshotManager = fileMergingSnapshotManager;
+ this.syncPhaseTimeoutMillis =
+ env.getJobConfiguration()
+
.get(CheckpointingOptions.CHECKPOINTING_SYNC_PHASE_TIMEOUT)
+ .toMillis();
+ if (syncPhaseTimeoutMillis > 0) {
+ LOG.info(
+ "Checkpoint synchronous phase timeout for task {} is
enabled with a timeout of {} ms.",
+ taskName,
+ syncPhaseTimeoutMillis);
+ }
}
public static ChannelStateWriter openChannelStateWriter(
@@ -747,6 +765,11 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
checkpointId, checkpointOptions.getTargetLocation());
storage = applyFileMergingCheckpoint(storage, checkpointOptions);
+ CountDownLatch syncPhaseCompleted = new CountDownLatch(1);
+ if (syncPhaseTimeoutMillis > 0) {
+ startSyncPhaseTimeoutWatchdog(checkpointId,
syncPhaseTimeoutMillis, syncPhaseCompleted);
+ }
+
try {
operatorChain.snapshotState(
operatorSnapshotsInProgress,
@@ -757,6 +780,7 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
storage);
} finally {
+ syncPhaseCompleted.countDown();
checkpointStorage.clearCacheFor(checkpointId);
}
@@ -849,4 +873,52 @@ class SubtaskCheckpointCoordinatorImpl implements
SubtaskCheckpointCoordinator {
delay);
}
}
+
+ private void startSyncPhaseTimeoutWatchdog(
+ long checkpointId, long syncPhaseTimeoutMillis, CountDownLatch
syncPhaseCompleted) {
+ final Thread taskThread = Thread.currentThread();
+ final Thread syncPhaseTimeoutWatchDog =
+ new Thread(
+ taskThread.getThreadGroup(),
+ () -> {
+ boolean completedInTime;
+ try {
+ completedInTime =
+ syncPhaseCompleted.await(
+ syncPhaseTimeoutMillis,
TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ // task shutdown
+ return;
+ }
+ if (completedInTime) {
+ return;
+ }
+ try {
+ final String errorMessage =
+ String.format(
+ "Task %s did not complete the
synchronous phase of checkpoint %s within %s ms.",
+ taskName, checkpointId,
syncPhaseTimeoutMillis);
+ if (LOG.isWarnEnabled()) {
+ LOG.warn(
+ errorMessage
+ + "\n"
+ +
Task.getTaskThreadStackTrace(taskThread));
+ }
+ env.failExternally(
+ new AsynchronousException(
+ new
TimeoutException(errorMessage)));
+ } catch (Exception e) {
+ LOG.error(
+ "Error handling sync phase timeout for
checkpoint {}",
+ checkpointId,
+ e);
+ }
+ },
+ String.format(
+ "checkpoint %s sync phase watchdog for %s_%s",
+ checkpointId, taskName,
env.getTaskInfo().getIndexOfThisSubtask()));
+ syncPhaseTimeoutWatchDog.setDaemon(true);
+
syncPhaseTimeoutWatchDog.setUncaughtExceptionHandler(FatalExitExceptionHandler.INSTANCE);
+ syncPhaseTimeoutWatchDog.start();
+ }
}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
index 6c161f139f6..574091204ef 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
@@ -18,9 +18,12 @@
package org.apache.flink.streaming.runtime.tasks;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.metrics.groups.OperatorMetricGroup;
@@ -73,6 +76,7 @@ import org.apache.flink.util.ExceptionUtils;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.CancellationException;
@@ -82,6 +86,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
@@ -95,6 +100,7 @@ import static org.assertj.core.api.Assertions.fail;
/** Tests for {@link SubtaskCheckpointCoordinator}. */
class SubtaskCheckpointCoordinatorTest {
private static final CheckpointStorage CHECKPOINT_STORAGE = new
JobManagerCheckpointStorage();
+ private static final long DEFAULT_SYNC_PHASE_TIMEOUT = 200L;
@Test
void testInitCheckpoint() throws IOException, CheckpointException {
@@ -378,6 +384,99 @@ class SubtaskCheckpointCoordinatorTest {
}
}
+ private StreamMockEnvironment getEnvWithCheckpointSyncTimeout() {
+ return new StreamMockEnvironment(
+ new Configuration(
+ new Configuration()
+ .set(
+
CheckpointingOptions.CHECKPOINTING_SYNC_PHASE_TIMEOUT,
+
Duration.ofMillis(DEFAULT_SYNC_PHASE_TIMEOUT))),
+ new Configuration(),
+ new ExecutionConfig(),
+ 1L,
+ new MockInputSplitProvider(),
+ 1,
+ new TestTaskStateManager());
+ }
+
+ @Test
+ void testSyncPhaseCompletedOnTime() throws Exception {
+ // Block for half the timeout so sync phase completes on time.
+ final long operatorBlockingUntil = DEFAULT_SYNC_PHASE_TIMEOUT / 2;
+ StreamMockEnvironment env = getEnvWithCheckpointSyncTimeout();
+ OneShotLatch unblockSnapshotLatch = new OneShotLatch();
+ BlockingCheckpointOperator operator =
+ new BlockingCheckpointOperator(unblockSnapshotLatch,
operatorBlockingUntil);
+
+ try (SubtaskCheckpointCoordinatorImpl subtaskCheckpointCoordinator =
+ (SubtaskCheckpointCoordinatorImpl)
+ new
MockSubtaskCheckpointCoordinatorBuilder().setEnvironment(env).build()) {
+ final OperatorChain<String, AbstractStreamOperator<String>>
operatorChain =
+ operatorChain(operator);
+
+ CheckpointOptions checkpointOptions =
+ new CheckpointOptions(
+ CheckpointType.CHECKPOINT,
+ CheckpointStorageLocationReference.getDefault(),
+ CheckpointOptions.AlignmentType.ALIGNED,
+ CheckpointOptions.NO_ALIGNED_CHECKPOINT_TIME_OUT);
+ subtaskCheckpointCoordinator.checkpointState(
+ new CheckpointMetaData(13L, System.currentTimeMillis()),
+ checkpointOptions,
+ new CheckpointMetricsBuilder()
+ .setAlignmentDurationNanos(0L)
+ .setBytesProcessedDuringAlignment(0L),
+ operatorChain,
+ false,
+ () -> true);
+ }
+ }
+
+ @Test
+ void testSyncPhaseWatchDogFailsStuckTask() throws Exception {
+ final long checkpointId = 13L;
+ // block for twice the timeout so sync phase times out
+ final long operatorBlockingUntil = DEFAULT_SYNC_PHASE_TIMEOUT * 2;
+ AtomicReference<Throwable> errorRef = new AtomicReference<>();
+ StreamMockEnvironment env = getEnvWithCheckpointSyncTimeout();
+ env.setExternalExceptionHandler(errorRef::set);
+ OneShotLatch unblockSnapshotLatch = new OneShotLatch();
+ BlockingCheckpointOperator operator =
+ new BlockingCheckpointOperator(unblockSnapshotLatch,
operatorBlockingUntil);
+
+ try (SubtaskCheckpointCoordinatorImpl subtaskCheckpointCoordinator =
+ (SubtaskCheckpointCoordinatorImpl)
+ new
MockSubtaskCheckpointCoordinatorBuilder().setEnvironment(env).build()) {
+ final OperatorChain<String, AbstractStreamOperator<String>>
operatorChain =
+ operatorChain(operator);
+
+ CheckpointOptions checkpointOptions =
+ new CheckpointOptions(
+ CheckpointType.CHECKPOINT,
+ CheckpointStorageLocationReference.getDefault(),
+ CheckpointOptions.AlignmentType.ALIGNED,
+ CheckpointOptions.NO_ALIGNED_CHECKPOINT_TIME_OUT);
+ subtaskCheckpointCoordinator.checkpointState(
+ new CheckpointMetaData(checkpointId,
System.currentTimeMillis()),
+ checkpointOptions,
+ new CheckpointMetricsBuilder(),
+ operatorChain,
+ false,
+ () -> true);
+ assertThat(errorRef.get()).isNotNull();
+ assertThat(errorRef.get().getCause())
+ .isInstanceOf(TimeoutException.class)
+ .hasMessageContaining(
+ "did not complete the synchronous phase of
checkpoint "
+ + checkpointId
+ + " within "
+ + DEFAULT_SYNC_PHASE_TIMEOUT
+ + " ms.");
+ } finally {
+ unblockSnapshotLatch.trigger();
+ }
+ }
+
@Test
void testBroadcastCancelCheckpointMarkerOnAbortingFromCoordinator() throws
Exception {
OneInputStreamTaskTestHarness<String, String> testHarness =
@@ -873,6 +972,33 @@ class SubtaskCheckpointCoordinatorTest {
public void processWatermarkStatus(WatermarkStatus watermarkStatus)
throws Exception {}
}
+ private static final class BlockingCheckpointOperator extends
CheckpointOperator {
+
+ private final OneShotLatch unblockSnapshotLatch;
+ private final long waitFor;
+
+ BlockingCheckpointOperator(OneShotLatch unblockSnapshotLatch, long
waitFor) {
+ super(new OperatorSnapshotFutures());
+ this.unblockSnapshotLatch = unblockSnapshotLatch;
+ this.waitFor = waitFor;
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory storageLocation)
+ throws Exception {
+ try {
+ unblockSnapshotLatch.await(waitFor, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException ignored) {
+
+ }
+ return super.snapshotState(checkpointId, timestamp,
checkpointOptions, storageLocation);
+ }
+ }
+
private static SubtaskCheckpointCoordinator coordinator(ChannelStateWriter
channelStateWriter)
throws IOException {
return new SubtaskCheckpointCoordinatorImpl(
diff --git
a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
index 064ceeccf69..b1c0cbc9150 100644
---
a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
+++
b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
@@ -163,6 +163,11 @@ public class TestStreamEnvironment extends
StreamExecutionEnvironment {
if (!conf.contains(CheckpointingOptions.FILE_MERGING_ENABLED)) {
randomize(conf, CheckpointingOptions.FILE_MERGING_ENABLED,
true);
}
+ randomize(
+ conf,
+ CheckpointingOptions.CHECKPOINTING_SYNC_PHASE_TIMEOUT,
+ CheckpointingOptions.CHECKPOINTING_TIMEOUT.defaultValue(),
+ Duration.ofMillis(0));
}
randomize(
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointSyncPhaseTimeoutITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointSyncPhaseTimeoutITCase.java
new file mode 100644
index 00000000000..5b9a27d2be8
--- /dev/null
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointSyncPhaseTimeoutITCase.java
@@ -0,0 +1,93 @@
+/*
+ * 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.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator;
+import org.apache.flink.streaming.util.RestartStrategyUtils;
+import org.apache.flink.test.junit5.InjectMiniCluster;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.test.util.InfiniteIntegerSource;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeoutException;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class CheckpointSyncPhaseTimeoutITCase extends TestLogger {
+
+ private static final long SYNC_PHASE_TIMEOUT_MILLIS = 50L;
+ private static final StreamExecutionEnvironment env = envSetup();
+
+ @RegisterExtension
+ static final MiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ private static StreamExecutionEnvironment envSetup() {
+ final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.enableCheckpointing(10);
+ env.getCheckpointConfig()
+
.setCheckpointSyncPhaseTimeout(Duration.ofMillis(SYNC_PHASE_TIMEOUT_MILLIS));
+ RestartStrategyUtils.configureNoRestartStrategy(env);
+ return env;
+ }
+
+ @Test
+ void testStuckSyncPhaseFailsJob(@InjectMiniCluster MiniCluster
miniCluster) throws Exception {
+ env.addSource(new BlockingSnapshotSource()).sinkTo(new
DiscardingSink<>());
+ JobGraph jobGraph =
StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
+
+ assertThatThrownBy(() -> miniCluster.executeJobBlocking(jobGraph))
+ .isInstanceOf(JobExecutionException.class)
+ .hasRootCauseInstanceOf(TimeoutException.class)
+ .hasRootCauseMessage(
+ "Task Source: Custom Source (1/1)#0 did not complete
the synchronous phase of checkpoint 1 within "
+ + SYNC_PHASE_TIMEOUT_MILLIS
+ + " ms.");
+ }
+
+ private static class BlockingSnapshotSource extends InfiniteIntegerSource
+ implements CheckpointedFunction {
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws
Exception {
+ new CountDownLatch(1).await();
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) {}
+ }
+}
