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)

Reply via email to