[FLINK-33080] The checkpoint storage configured in the job level by config option will not take effect

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.18.0, 1.17.1, 1.19.0
• Fix Version/s: 1.19.0
• Component/s: Runtime / Checkpointing, Runtime / Configuration
• Labels: pull-request-available
• Assignee: Junrui Li
• Reporter: Junrui Li
• Created: 13/Sep/23 07:02
• Updated: 31/Oct/23 03:38
• Resolved: 31/Oct/23 03:38

Description

When we configure the checkpoint storage at the job level, it can only be done through the following method:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getCheckpointConfig().setCheckpointStorage(xxx); 

or configure filesystem storage by config option CheckpointingOptions.CHECKPOINTS_DIRECTORY through the following method:

Configuration configuration = new Configuration();
configuration.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir); 
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

However, configure the other type checkpoint storage by the job-side configuration like the following will not take effect:

Configuration configuration = new Configuration();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
configuration.set(CheckpointingOptions.CHECKPOINT_STORAGE, "aaa.bbb.ccc.CustomCheckpointStorage");
configuration.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir); 

This behavior is unexpected, we should allow this way will take effect.

Issue Links

• links to: GitHub Pull Request #23408