This is an automated email from the ASF dual-hosted git repository.
MartijnVisser 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 353c25629ca [FLINK-40916][runtime] Add option to create checkpoint
directories on first checkpoint (#29409)
353c25629ca is described below
commit 353c25629cad73afcf74c0d098ae3173b9f35962
Author: Isaac Montenegro Jimenez <[email protected]>
AuthorDate: Tue Oct 20 20:00:34 2026 +0100
[FLINK-40916][runtime] Add option to create checkpoint directories on first
checkpoint (#29409)
---
.../generated/checkpointing_configuration.html | 6 +
.../generated/expert_checkpointing_section.html | 6 +
.../flink/configuration/CheckpointingOptions.java | 24 ++++
.../runtime/checkpoint/CheckpointCoordinator.java | 23 ++--
.../tasks/CheckpointCoordinatorConfiguration.java | 31 ++++-
.../flink/streaming/api/graph/StreamGraph.java | 3 +
.../checkpoint/CheckpointCoordinatorTest.java | 151 +++++++++++++++++++++
.../api/graph/StreamGraphGeneratorTest.java | 22 +++
8 files changed, 251 insertions(+), 15 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/checkpointing_configuration.html
b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
index c256e4a66e6..e5fb53d40ca 100644
--- a/docs/layouts/shortcodes/generated/checkpointing_configuration.html
+++ b/docs/layouts/shortcodes/generated/checkpointing_configuration.html
@@ -26,6 +26,12 @@
<td>Boolean</td>
<td>Option whether to discard a checkpoint's states in parallel
using the ExecutorService passed into the cleaner</td>
</tr>
+ <tr>
+
<td><h5>execution.checkpointing.create-directories-on-job-start</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to create the directories of the checkpoint storage
(for example the 'shared' and 'taskowned' directories) when the job starts. If
set to false, they are created when the first checkpoint is triggered instead,
which avoids blocking job startup on file systems where creating directories is
a slow remote call, such as object stores.<br />WARNING: If set to false, an
unusable checkpoint location is no longer detected when the job starts. It
surfaces as checkpoint fai [...]
+ </tr>
<tr>
<td><h5>execution.checkpointing.create-subdir</h5></td>
<td style="word-wrap: break-word;">true</td>
diff --git
a/docs/layouts/shortcodes/generated/expert_checkpointing_section.html
b/docs/layouts/shortcodes/generated/expert_checkpointing_section.html
index 582a61f89f3..cca09d262a4 100644
--- a/docs/layouts/shortcodes/generated/expert_checkpointing_section.html
+++ b/docs/layouts/shortcodes/generated/expert_checkpointing_section.html
@@ -8,6 +8,12 @@
</tr>
</thead>
<tbody>
+ <tr>
+
<td><h5>execution.checkpointing.create-directories-on-job-start</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to create the directories of the checkpoint storage
(for example the 'shared' and 'taskowned' directories) when the job starts. If
set to false, they are created when the first checkpoint is triggered instead,
which avoids blocking job startup on file systems where creating directories is
a slow remote call, such as object stores.<br />WARNING: If set to false, an
unusable checkpoint location is no longer detected when the job starts. It
surfaces as checkpoint fai [...]
+ </tr>
<tr>
<td><h5>execution.checkpointing.create-subdir</h5></td>
<td style="word-wrap: break-word;">true</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 b36f9767ed4..2c9f3834177 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
@@ -231,6 +231,30 @@ public class CheckpointingOptions {
+ "restoration of the
current job when starting a new job.")
.build());
+ /**
+ * Whether the checkpoint directories are created when the job starts,
instead of when the first
+ * checkpoint is triggered.
+ */
+ @Documentation.Section(Documentation.Sections.EXPERT_CHECKPOINTING)
+ public static final ConfigOption<Boolean> CREATE_DIRECTORIES_ON_JOB_START =
+
ConfigOptions.key("execution.checkpointing.create-directories-on-job-start")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Whether to create the directories
of the checkpoint storage (for example the "
+ + "'shared' and
'taskowned' directories) when the job starts. If set to false, "
+ + "they are created when
the first checkpoint is triggered instead, which avoids "
+ + "blocking job startup on
file systems where creating directories is a slow "
+ + "remote call, such as
object stores.")
+ .linebreak()
+ .text(
+ "WARNING: If set to false, an
unusable checkpoint location is no longer detected "
+ + "when the job starts. It
surfaces as checkpoint failures, and fails the job "
+ + "once the tolerable
number of checkpoint failures is exceeded.")
+ .build());
+
/**
* The minimum size of state data files. All state chunks smaller than
that are stored inline in
* the root checkpoint metadata file.
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index 7750543be41..c22ee6ffe4c 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.core.execution.CheckpointType;
import org.apache.flink.core.execution.SavepointFormatType;
import
org.apache.flink.runtime.checkpoint.FinishedTaskStateProvider.PartialFinishingNotSupportedByStateException;
@@ -253,7 +254,8 @@ public class CheckpointCoordinator {
@GuardedBy("lock")
private final Set<OperatorID> backlogOperators = new HashSet<>();
- private boolean baseLocationsForCheckpointInitialized = false;
+ // Written on the IO executor.
+ private volatile boolean baseLocationsForCheckpointInitialized = false;
private boolean forceFullSnapshot;
@@ -351,7 +353,12 @@ public class CheckpointCoordinator {
try {
this.checkpointStorageView =
checkpointStorage.createCheckpointStorage(job);
- if (isPeriodicCheckpointingConfigured()) {
+ if (!chkConfig.isCreateDirectoriesOnJobStart()) {
+ LOG.info(
+ "Checkpoint directories for job {} will be created
when the first checkpoint is triggered ({}=false).",
+ job,
+
CheckpointingOptions.CREATE_DIRECTORIES_ON_JOB_START.key());
+ } else if (isPeriodicCheckpointingConfigured()) {
checkpointStorageView.initializeBaseLocationsForCheckpoint();
baseLocationsForCheckpointInitialized = true;
}
@@ -637,9 +644,6 @@ public class CheckpointCoordinator {
CompletableFuture<CheckpointPlan> checkpointPlanFuture =
checkpointPlanCalculator.calculateCheckpointPlan();
- boolean initializeBaseLocations =
!baseLocationsForCheckpointInitialized;
- baseLocationsForCheckpointInitialized = true;
-
CompletableFuture<Void> masterTriggerCompletionPromise = new
CompletableFuture<>();
final CompletableFuture<PendingCheckpoint>
pendingCheckpointCompletableFuture =
@@ -679,8 +683,7 @@ public class CheckpointCoordinator {
initializeCheckpointLocation(
pendingCheckpoint.getCheckpointID(),
request.props,
-
request.externalSavepointLocation,
-
initializeBaseLocations);
+
request.externalSavepointLocation);
return Tuple2.of(
pendingCheckpoint,
checkpointStorageLocation);
} catch (Throwable e) {
@@ -874,8 +877,7 @@ public class CheckpointCoordinator {
private CheckpointStorageLocation initializeCheckpointLocation(
long checkpointID,
CheckpointProperties props,
- @Nullable String externalSavepointLocation,
- boolean initializeBaseLocations)
+ @Nullable String externalSavepointLocation)
throws Exception {
final CheckpointStorageLocation checkpointStorageLocation;
if (props.isSavepoint()) {
@@ -883,8 +885,9 @@ public class CheckpointCoordinator {
checkpointStorageView.initializeLocationForSavepoint(
checkpointID, externalSavepointLocation);
} else {
- if (initializeBaseLocations) {
+ if (!baseLocationsForCheckpointInitialized) {
checkpointStorageView.initializeBaseLocationsForCheckpoint();
+ baseLocationsForCheckpointInitialized = true;
}
checkpointStorageLocation =
checkpointStorageView.initializeLocationForCheckpoint(checkpointID);
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java
index 1f4702213e4..bd7a4e0dc29 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointCoordinatorConfiguration.java
@@ -76,6 +76,8 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
private final boolean pauseSourcesUntilFirstCheckpoint;
+ private final boolean createDirectoriesOnJobStart;
+
/**
* @deprecated use {@link #builder()}.
*/
@@ -105,7 +107,8 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
checkpointIdOfIgnoredInFlightData,
false,
false,
- false);
+ false,
+ true);
}
private CheckpointCoordinatorConfiguration(
@@ -122,7 +125,8 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
long checkpointIdOfIgnoredInFlightData,
boolean enableCheckpointsAfterTasksFinish,
boolean recoverOutputOnDownstreamTask,
- boolean pauseSourcesUntilFirstCheckpoint) {
+ boolean pauseSourcesUntilFirstCheckpoint,
+ boolean createDirectoriesOnJobStart) {
if (checkpointIntervalDuringBacklog < MINIMAL_CHECKPOINT_TIME) {
// interval of max value means disable periodic checkpoint
@@ -164,6 +168,7 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
this.enableCheckpointsAfterTasksFinish =
enableCheckpointsAfterTasksFinish;
this.recoverOutputOnDownstreamTask = recoverOutputOnDownstreamTask;
this.pauseSourcesUntilFirstCheckpoint =
pauseSourcesUntilFirstCheckpoint;
+ this.createDirectoriesOnJobStart = createDirectoriesOnJobStart;
}
public long getCheckpointInterval() {
@@ -222,6 +227,10 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
return recoverOutputOnDownstreamTask;
}
+ public boolean isCreateDirectoriesOnJobStart() {
+ return createDirectoriesOnJobStart;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -242,7 +251,8 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
&& tolerableCheckpointFailureNumber ==
that.tolerableCheckpointFailureNumber
&& checkpointIdOfIgnoredInFlightData ==
that.checkpointIdOfIgnoredInFlightData
&& enableCheckpointsAfterTasksFinish ==
that.enableCheckpointsAfterTasksFinish
- && recoverOutputOnDownstreamTask ==
that.recoverOutputOnDownstreamTask;
+ && recoverOutputOnDownstreamTask ==
that.recoverOutputOnDownstreamTask
+ && createDirectoriesOnJobStart ==
that.createDirectoriesOnJobStart;
}
@Override
@@ -259,7 +269,8 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
tolerableCheckpointFailureNumber,
checkpointIdOfIgnoredInFlightData,
enableCheckpointsAfterTasksFinish,
- recoverOutputOnDownstreamTask);
+ recoverOutputOnDownstreamTask,
+ createDirectoriesOnJobStart);
}
@Override
@@ -289,6 +300,8 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
+ enableCheckpointsAfterTasksFinish
+ ", recoverOutputOnDownstreamTask="
+ recoverOutputOnDownstreamTask
+ + ", createDirectoriesOnJobStart="
+ + createDirectoriesOnJobStart
+ '}';
}
@@ -332,6 +345,7 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
private boolean enableCheckpointsAfterTasksFinish;
private boolean recoverOutputOnDownstreamTask;
private boolean pauseSourcesUntilFirstCheckpoint;
+ private boolean createDirectoriesOnJobStart = true;
public CheckpointCoordinatorConfiguration build() {
return new CheckpointCoordinatorConfiguration(
@@ -348,7 +362,8 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
checkpointIdOfIgnoredInFlightData,
enableCheckpointsAfterTasksFinish,
recoverOutputOnDownstreamTask,
- pauseSourcesUntilFirstCheckpoint);
+ pauseSourcesUntilFirstCheckpoint,
+ createDirectoriesOnJobStart);
}
public CheckpointCoordinatorConfigurationBuilder setCheckpointInterval(
@@ -433,5 +448,11 @@ public class CheckpointCoordinatorConfiguration implements
Serializable {
this.recoverOutputOnDownstreamTask = recoverOutputOnDownstreamTask;
return this;
}
+
+ public CheckpointCoordinatorConfigurationBuilder
setCreateDirectoriesOnJobStart(
+ boolean createDirectoriesOnJobStart) {
+ this.createDirectoriesOnJobStart = createDirectoriesOnJobStart;
+ return this;
+ }
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index 6d2e112d6c6..702fa773de6 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -401,6 +401,9 @@ public class StreamGraph implements Pipeline, ExecutionPlan
{
.UNALIGNED_RECOVER_OUTPUT_ON_DOWNSTREAM))
.setPauseSourcesUntilFirstCheckpoint(
cfg.isPauseSourcesUntilFirstCheckpoint())
+ .setCreateDirectoriesOnJobStart(
+ jobConfiguration.get(
+
CheckpointingOptions.CREATE_DIRECTORIES_ON_JOB_START))
.build(),
serializedStateBackend,
getJobConfiguration()
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index 2b5e6653dc3..16a36b7d7bd 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -68,6 +68,7 @@ import org.apache.flink.runtime.state.SharedStateRegistryImpl;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.TestingStreamStateHandle;
import org.apache.flink.runtime.state.filesystem.FileStateHandle;
+import org.apache.flink.runtime.state.filesystem.FsCheckpointStorageAccess;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
import
org.apache.flink.runtime.state.memory.MemoryBackendCheckpointStorageAccess;
import
org.apache.flink.runtime.state.memory.NonPersistentMetadataCheckpointStorageLocation;
@@ -123,6 +124,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -4049,6 +4051,121 @@ class CheckpointCoordinatorTest {
assertThat(fs.exists(jobCheckpointPath)).isFalse();
}
+ @Test
+ void testBaseLocationsInitializedOnJobStartNotRepeatedOnTrigger() throws
Exception {
+ BaseLocationsTrackingCheckpointStorage storage =
+ new
BaseLocationsTrackingCheckpointStorage(TempDirUtils.newFolder(tmpFolder), 0);
+ CheckpointCoordinator checkpointCoordinator =
+ new CheckpointCoordinatorBuilder()
+ .setCheckpointStorage(storage)
+ .setTimer(manuallyTriggeredScheduledExecutor)
+ .build(EXECUTOR_RESOURCE.getExecutor());
+
+
assertThat(checkpointCoordinator.isPeriodicCheckpointingConfigured()).isTrue();
+ assertThat(storage.getInitializeBaseLocationsCalls()).isOne();
+
+ triggerCheckpointAndRunTimer(checkpointCoordinator);
+ assertThat(storage.getInitializeBaseLocationsCalls()).isOne();
+
+ checkpointCoordinator.shutdown();
+ }
+
+ @Test
+ void
testBaseLocationsInitializedOnFirstCheckpointWhenCreateDirectoriesOnJobStartDisabled()
+ throws Exception {
+ BaseLocationsTrackingCheckpointStorage storage =
+ new
BaseLocationsTrackingCheckpointStorage(TempDirUtils.newFolder(tmpFolder), 0);
+ CheckpointCoordinator checkpointCoordinator =
+ new CheckpointCoordinatorBuilder()
+ .setCheckpointCoordinatorConfiguration(
+ CheckpointCoordinatorConfiguration.builder()
+
.setMaxConcurrentCheckpoints(Integer.MAX_VALUE)
+ .setCreateDirectoriesOnJobStart(false)
+ .build())
+ .setCheckpointStorage(storage)
+ .setTimer(manuallyTriggeredScheduledExecutor)
+ .build(EXECUTOR_RESOURCE.getExecutor());
+
+
assertThat(checkpointCoordinator.isPeriodicCheckpointingConfigured()).isTrue();
+ assertThat(storage.getInitializeBaseLocationsCalls()).isZero();
+
+ triggerCheckpointAndRunTimer(checkpointCoordinator);
+
assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isOne();
+ assertThat(storage.getInitializeBaseLocationsCalls()).isOne();
+
+ triggerCheckpointAndRunTimer(checkpointCoordinator);
+
assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isEqualTo(2);
+ assertThat(storage.getInitializeBaseLocationsCalls()).isOne();
+
+ checkpointCoordinator.shutdown();
+ }
+
+ @Test
+ void testBaseLocationsInitializedOnFirstCheckpointAfterSavepoint() throws
Exception {
+ BaseLocationsTrackingCheckpointStorage storage =
+ new
BaseLocationsTrackingCheckpointStorage(TempDirUtils.newFolder(tmpFolder), 0);
+ CheckpointCoordinator checkpointCoordinator =
+ buildCoordinatorWithoutPeriodicCheckpointing(storage);
+
+ String savepointDir =
TempDirUtils.newFolder(tmpFolder).getAbsolutePath();
+ checkpointCoordinator.triggerSavepoint(savepointDir,
SavepointFormatType.CANONICAL);
+ manuallyTriggeredScheduledExecutor.triggerAll();
+
assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isOne();
+ assertThat(storage.getInitializeBaseLocationsCalls()).isZero();
+
+ triggerCheckpointAndRunTimer(checkpointCoordinator);
+
assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isEqualTo(2);
+ assertThat(storage.getInitializeBaseLocationsCalls()).isOne();
+
+ checkpointCoordinator.shutdown();
+ }
+
+ @Test
+ void
testBaseLocationsInitializationRetriedAfterFailureAndNotRepeatedAfterSuccess()
+ throws Exception {
+ BaseLocationsTrackingCheckpointStorage storage =
+ new
BaseLocationsTrackingCheckpointStorage(TempDirUtils.newFolder(tmpFolder), 1);
+ CheckpointCoordinator checkpointCoordinator =
+ buildCoordinatorWithoutPeriodicCheckpointing(storage);
+
+ CompletableFuture<CompletedCheckpoint> failedCheckpoint =
+ triggerCheckpointAndRunTimer(checkpointCoordinator);
+ assertThat(failedCheckpoint).isCompletedExceptionally();
+
assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isZero();
+ assertThat(storage.getInitializeBaseLocationsCalls()).isOne();
+
+ triggerCheckpointAndRunTimer(checkpointCoordinator);
+
assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isOne();
+ assertThat(storage.getInitializeBaseLocationsCalls()).isEqualTo(2);
+
+ triggerCheckpointAndRunTimer(checkpointCoordinator);
+
assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isEqualTo(2);
+ assertThat(storage.getInitializeBaseLocationsCalls()).isEqualTo(2);
+
+ checkpointCoordinator.shutdown();
+ }
+
+ private CheckpointCoordinator buildCoordinatorWithoutPeriodicCheckpointing(
+ CheckpointStorage storage) throws Exception {
+ return new CheckpointCoordinatorBuilder()
+ .setCheckpointCoordinatorConfiguration(
+ CheckpointCoordinatorConfiguration.builder()
+ .setCheckpointInterval(Long.MAX_VALUE)
+ .setMaxConcurrentCheckpoints(Integer.MAX_VALUE)
+ .build())
+ .setCheckpointStorage(storage)
+ .setTimer(manuallyTriggeredScheduledExecutor)
+ .build(EXECUTOR_RESOURCE.getExecutor());
+ }
+
+ private CompletableFuture<CompletedCheckpoint>
triggerCheckpointAndRunTimer(
+ CheckpointCoordinator checkpointCoordinator) {
+ CompletableFuture<CompletedCheckpoint> checkpoint =
+ checkpointCoordinator.triggerCheckpoint(false);
+ manuallyTriggeredScheduledExecutor.triggerAll();
+ return checkpoint;
+ }
+
private CheckpointCoordinator getCheckpointCoordinator(ExecutionGraph
graph) throws Exception {
return new CheckpointCoordinatorBuilder()
.setCheckpointCoordinatorConfiguration(
@@ -4220,6 +4337,40 @@ class CheckpointCoordinatorTest {
}
}
+ /** Counts base location initializations, failing the first {@code
failuresBeforeSuccess}. */
+ private static class BaseLocationsTrackingCheckpointStorage
+ extends FileSystemCheckpointStorage {
+ private final AtomicInteger initializeBaseLocationsCalls = new
AtomicInteger();
+ private final int failuresBeforeSuccess;
+
+ BaseLocationsTrackingCheckpointStorage(File checkpointDir, int
failuresBeforeSuccess) {
+ super(checkpointDir.toURI());
+ this.failuresBeforeSuccess = failuresBeforeSuccess;
+ }
+
+ int getInitializeBaseLocationsCalls() {
+ return initializeBaseLocationsCalls.get();
+ }
+
+ @Override
+ public CheckpointStorageAccess createCheckpointStorage(JobID jobId)
throws IOException {
+ return new FsCheckpointStorageAccess(
+ getCheckpointPath(),
+ getSavepointPath(),
+ jobId,
+ getMinFileSizeThreshold(),
+ getWriteBufferSize()) {
+ @Override
+ public void initializeBaseLocationsForCheckpoint() throws
IOException {
+ if (initializeBaseLocationsCalls.incrementAndGet() <=
failuresBeforeSuccess) {
+ throw new IOException("Failed to initialize base
locations");
+ }
+ super.initializeBaseLocationsForCheckpoint();
+ }
+ };
+ }
+ }
+
private static class StoppingCheckpointIDCounter extends
CheckpointIDCounterWithOwner {
@Override
public long getAndIncrement() throws Exception {
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
index dbde71ef687..6384cc60f19 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
@@ -27,11 +27,13 @@ import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.core.memory.ManagedMemoryUseCase;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.CachedDataStream;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
@@ -86,6 +88,8 @@ import org.hamcrest.FeatureMatcher;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeMatcher;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Arrays;
@@ -1255,4 +1259,22 @@ class StreamGraphGeneratorTest {
return new HashSet<>(completedClusterDatasetIds);
}
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void
testCreateDirectoriesOnJobStartIsForwardedToCheckpointSettings(boolean
createOnJobStart)
+ throws Exception {
+ Configuration configuration = new Configuration();
+
configuration.set(CheckpointingOptions.CREATE_DIRECTORIES_ON_JOB_START,
createOnJobStart);
+ StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ env.enableCheckpointing(1000);
+ env.fromData(1, 2, 3).sinkTo(new DiscardingSink<>());
+
+ JobCheckpointingSettings settings =
+ env.getStreamGraph().getJobGraph().getCheckpointingSettings();
+
+
assertThat(settings.getCheckpointCoordinatorConfiguration().isCreateDirectoriesOnJobStart())
+ .isEqualTo(createOnJobStart);
+ }
}
