Matthias created FLINK-24506: -------------------------------- Summary: checkpoint directory is not configurable through the Flink configuration passed into the StreamExecutionEnvironment Key: FLINK-24506 URL: https://issues.apache.org/jira/browse/FLINK-24506 Project: Flink Issue Type: Bug Components: Runtime / Configuration, Runtime / State Backends Affects Versions: 1.13.2, 1.14.0 Reporter: Matthias
FLINK-19463 introduced the separation of `StateBackend` and `CheckpointStorage`. Before that, both were included in the same interface implementation [AbstractFileStateBackend|https://github.com/apache/flink/blob/0a76daba0a428a322f0273d7dc6a70966f62bf26/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/AbstractFileStateBackend.java]. `FsStateBackend` was used as a default implementation pre-1.13. pre-{{1.13}} initialized the checkpoint directory when instantiating the state backend (see [FsStateBackendFactory|https://github.com/apache/flink/blob/release-1.12/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsStateBackendFactory.java#L46]). Starting from {{1.13}} loading the {{CheckpointStorage}} is done by the {{CheckpointStorageLoader.load}} method that is called in various places: * Savepoint Disposal (through {{Checkpoints.loadCheckpointStorage}}) where it only relies on the configuration passed in by the cluster configuration (no application checkpoint storage is passed) * {{SchedulerBase}} initialization (through DefaultExecutionGraphBuilder) where it’s based on the cluster’s configuration but also the application configuration (i.e. the {{JobGraph}}’s setting) that would be considered if {{CheckpointConfig#configure}} would have the checkpoint storage included * {{StreamTask}} on the {{TaskManager}}’s side where it’s based on the configuration passed in by the {{JobVertex}} for the application’s {{CheckpointStorage}} and the {[TaskManager}}’s configuration (coming from the session cluster) for the fallback {{CheckpointStorage}} The issue is that we don't set the checkpoint directory in the {{CheckpointConfig}}. Hence, it's not going to get picked up as a job-related property. Flink always uses the fallback provided by the session cluster configuration. -- This message was sent by Atlassian Jira (v8.3.4#803005)