rkhachatryan commented on code in PR #28786:
URL: https://github.com/apache/flink/pull/28786#discussion_r3730954918
##########
flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java:
##########
@@ -583,6 +583,14 @@ 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 blocking part of a checkpoint,
after which a job failure is triggered "
+ + "to address thread blockages. Defaults
to 0 (disabled).");
Review Comment:
Could you also describe what happens further - especially if the task thread
doesn't unblock?
IIUC: task cancellation -> task cancellation timeout -> TM shutdown (and
potentially start-up)
##########
flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java:
##########
@@ -165,6 +165,11 @@ private static void randomizeConfiguration(MiniCluster
miniCluster, Configuratio
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));
Review Comment:
Are the values intentional here?
We set sync phase timeout to 10 minutes in all the tests.
If some test sets CHECKPOINTING_TIMEOUT to a different value, we might
trigger sync phase timeout. Although 10-minutes timeout in test sounds
unrealistic.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java:
##########
@@ -1879,20 +1879,23 @@ public void run() {
}
}
- public static void logTaskThreadStackTrace(
- Thread thread, String taskName, long timeoutMs, String action) {
+ public static String getTaskThreadStackTrace(Thread thread) {
StackTraceElement[] stack = thread.getStackTrace();
StringBuilder stackTraceStr = new StringBuilder();
for (StackTraceElement e : stack) {
stackTraceStr.append(e).append('\n');
}
+ return stackTraceStr.toString();
+ }
+ 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));
Review Comment:
Can we guard this computation as well (if LOG.isWarn enabled)?
It now invoked `stackTraceStr.toString();` unconditionally.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java:
##########
@@ -849,4 +865,51 @@ private static void
logCheckpointProcessingDelay(CheckpointMetaData checkpointMe
delay);
}
}
+
+ private Thread startSyncPhaseTimeoutWatchdog(
+ long checkpointId, long syncPhaseTimeoutMillis, CountDownLatch
syncPhaseCompleted) {
+ final Thread taskThread = Thread.currentThread();
+ final Thread syncPhaseTimeoutWatchDog =
+ new Thread(
+ taskThread.getThreadGroup(),
+ () -> {
+ try {
+ syncPhaseCompleted.await(
+ syncPhaseTimeoutMillis,
TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ // task shutdown
+ return;
+ }
+ if (syncPhaseCompleted.getCount() == 0) {
+ return;
+ }
Review Comment:
NIT: can be replaced by checking the return value of `await` above.
##########
flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointSyncPhaseTimeoutITCase.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.TimeUnit;
+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(2 * SYNC_PHASE_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS);
Review Comment:
Why not just block forever?
IMO that'd be both more realistic and deterministic
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java:
##########
@@ -849,4 +865,51 @@ private static void
logCheckpointProcessingDelay(CheckpointMetaData checkpointMe
delay);
}
}
+
+ private Thread startSyncPhaseTimeoutWatchdog(
Review Comment:
NIT: return value never used
##########
flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java:
##########
@@ -583,6 +583,14 @@ 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 blocking part of a checkpoint,
after which a job failure is triggered "
Review Comment:
NITs:
`blocking part` => `synchronous phase`
`job failure` => `task failure`
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java:
##########
@@ -849,4 +865,51 @@ private static void
logCheckpointProcessingDelay(CheckpointMetaData checkpointMe
delay);
}
}
+
+ private Thread startSyncPhaseTimeoutWatchdog(
+ long checkpointId, long syncPhaseTimeoutMillis, CountDownLatch
syncPhaseCompleted) {
+ final Thread taskThread = Thread.currentThread();
+ final Thread syncPhaseTimeoutWatchDog =
+ new Thread(
+ taskThread.getThreadGroup(),
+ () -> {
+ try {
+ syncPhaseCompleted.await(
+ syncPhaseTimeoutMillis,
TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ // task shutdown
+ return;
+ }
+ if (syncPhaseCompleted.getCount() == 0) {
+ 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()));
Review Comment:
Can this exceed the maximum length of thread name? 🤔
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java:
##########
@@ -747,6 +753,15 @@ private boolean takeSnapshotSync(
checkpointId, checkpointOptions.getTargetLocation());
storage = applyFileMergingCheckpoint(storage, checkpointOptions);
+ final long syncPhaseTimeoutMillis =
+ env.getJobConfiguration()
+
.get(CheckpointingOptions.CHECKPOINTING_SYNC_PHASE_TIMEOUT)
+ .toMillis();
+ CountDownLatch syncPhaseCompleted = new CountDownLatch(1);
+ if (syncPhaseTimeoutMillis > 0) {
+ startSyncPhaseTimeoutWatchdog(checkpointId,
syncPhaseTimeoutMillis, syncPhaseCompleted);
+ }
Review Comment:
Can we log the chosen behavior? Maybe extract this to a field and log in the
constructor?
--
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]
