diff --git a/docs/layouts/shortcodes/generated/checkpointing_configuration.html b/docs/layouts/shortcodes/generated/checkpointing_configuration.html index 2fcbc5fc1edb54..524c40ddb0163e 100644 --- a/docs/layouts/shortcodes/generated/checkpointing_configuration.html +++ b/docs/layouts/shortcodes/generated/checkpointing_configuration.html @@ -152,6 +152,12 @@ String The checkpoint storage implementation to be used to checkpoint state.
The implementation can be specified either via their shortcut name, or via the class name of a CheckpointStorageFactory. If a factory is specified it is instantiated via its zero argument constructor and its CheckpointStorageFactory#createFromConfig(ReadableConfig, ClassLoader) method is called.
Recognized shortcut names are 'jobmanager' and 'filesystem'.
'execution.checkpointing.storage' and 'execution.checkpointing.dir' are usually combined to configure the checkpoint location. By default, the checkpoint meta data and actual program state will be stored in the JobManager's memory directly. When 'execution.checkpointing.storage' is set to 'jobmanager', if 'execution.checkpointing.dir' is configured, the meta data of checkpoints will be persisted to the path specified by 'execution.checkpointing.dir'. Otherwise, the meta data will be stored in the JobManager's memory. When 'execution.checkpointing.storage' is set to 'filesystem', a valid path must be configured to 'execution.checkpointing.dir', and the checkpoint meta data and actual program state will both be persisted to the path. + +
execution.checkpointing.sync-phase-timeout
+ 0 ms + Duration + A timeout for the blocking part of a checkpoint, after which a job failure is triggered to address thread blockages. Defaults to 0 (disabled). +
execution.checkpointing.timeout
10 min 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 75abd4178a0ca6..ed12a6ab6346db 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,14 @@ public class CheckpointingOptions { + "will timeout and checkpoint barrier will start working as unaligned checkpoint.") .build()); + public static final ConfigOption 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)."); + public static final ConfigOption 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 249f29af1be66e..330adef9e1ba62 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 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)); } /** 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 8d0982a772d72b..88d47fd245d174 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 void setAlignedCheckpointTimeout(Duration alignedCheckpointTimeout) { 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 void configure(ReadableConfig configuration) { 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 811354c47f380a..e0472d5994cfb7 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.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.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.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; @@ -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); + } + try { operatorChain.snapshotState( operatorSnapshotsInProgress, @@ -757,6 +772,7 @@ private boolean takeSnapshotSync( storage); } finally { + syncPhaseCompleted.countDown(); checkpointStorage.clearCacheFor(checkpointId); } @@ -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())); + syncPhaseTimeoutWatchDog.setDaemon(true); + syncPhaseTimeoutWatchDog.setUncaughtExceptionHandler(FatalExitExceptionHandler.INSTANCE); + syncPhaseTimeoutWatchDog.start(); + return syncPhaseTimeoutWatchDog; + } } 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 6c161f139f68ac..574091204ef03b 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.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.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 @@ /** 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 @@ void testNotifyCheckpointAbortedBeforeAsyncPhase() throws Exception { } } + 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> 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 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> 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 testHarness = @@ -873,6 +972,33 @@ public void processLatencyMarker(LatencyMarker latencyMarker) {} 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 c51108d7ee2ac4..801f52e9341fb7 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 @@ -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)); } 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 00000000000000..9389bea00b150d --- /dev/null +++ b/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); + } + + @Override + public void initializeState(FunctionInitializationContext context) {} + } +}