...

...

Discussion thread: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-FLIP-59-Enable-execution-configuration-from-Configuration-object-td32359.html

JIRA:

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

This FLIP in few places is based on FLIP-54. In particular it assumes richer type support for ConfigOptions, and availability of ConfigurationReader#getOptional.

...

Proposed Changes

Methods to configure config objects

We suggest to add methods that enable to change the state of the three classes from a ConfigurationReader.

• Add configure methods:
• void ExecutionConfig.configure(ConfigurationReaderReadableConfig configuration, ClassLoader classLoader)
• void CheckpointConfig.configure(ConfigurationReaderReadableConfig)
• void StreamExecutionEnvironment.configure(ConfigurationReaderReadableConfig configuration, ClassLoader classLoader)

    Those methods would mutate internal state of only those parameters for which a corresponding option is present in the ConfigurationReader. The rest would be left intact. The method StreamExecutionEnvironment.configure would also call the two other methods.

E.g.

    public static final ConfigOption<ClosureCleanerLevel> CLOSURE_CLEANER_LEVEL =

        ConfigOptions.key("execpipeline.closure-cleaner-level")

            .enumType(ClosureCleanerLevel.class)

            .defaultValue(ClosureCleanerLevel.RECURSIVE);


    void configure(ConfigurationReader conf) {

             ...

            conf.getOptional(CLOSURE_CLEANER_LEVEL).ifPresent(level -> {

                 this.closureCleanerLevel = level;

            })

            ...

    }

• Add three ConfigOptionSets:
• JobOptions in flink-core (this will include all options from ExecutionConfig + bufferTimeout, isChainingEnabled, CacheFile from StreamExecutionEnvironment) 
• CheckpointJobOptions in flink-streaming-java (we do not want to use the CheckpointingOptions to better differentiate from cluster-side options + CheckpointingMode & ExternalizedCheckpointCleanup are not available in flink-core)
• StreamJobOptions in flink-streaming-java (for TimeCharacteristic) - at some point in the future we could merge with JobOptions, once we resolve dependencies (TimeCharacteristic is not in flink-core)

Proposed Changes

...

New ConfigOptions

• In class org.apache.flink.configuration.PipelineOptions:

org.apache.flink.api.common.ExecutionConfigTest;org.apache.flink.api.common.ExecutionConfigTest$TestSerializer1

org.apache.flink.api.common.ExecutionConfigTest;org.apache.flink.api.common.ExecutionConfigTest$TestSerializer1

name:file1,path:/tmp1,executable:true;name:file2,path:/tmp2

• In class org.apache.flink.streaming.api.environment.StreamPipelineOptions:

  key	setting	type
  pipeline.time-characteristic	StreamExecutionEnvironment#timeCharacteristic	ConfigOption<TimeCharacteristic>

  
  

• In class org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions

  key	setting	type
  execution.checkpointing.mode	CheckpointConfig#checkpointingMode	ConfigOption<CheckpointingMode>
  execution.checkpointing.interval	CheckpointConfig#checkpointInterval	ConfigOption<Duration>
  execution.checkpointing.timeout	CheckpointConfig#checkpointTimeout	ConfigOption<Duration>
  execution.checkpointing.max-concurrent-checkpoints	CheckpointConfig#maxConcurrentCheckpoints	ConfigOption<Integer>
  execution.checkpointing.min-pause	CheckpointConfig#minPauseBetweenCheckpoints	ConfigOption<Duration>
  execution.checkpointing.prefer-checkpoint-for-recovery	CheckpointConfig#preferCheckpointForRecovery	ConfigOption<Boolean>
  execution.checkpointing.tolerable-failed-checkpoints	CheckpointConfig#tolerableCheckpointFailureNumber	ConfigOption<Integer>
  execution.checkpointing.externalized-checkpoint	CheckpointConfig#externalizedCheckpointCleanup	ConfigOption<CheckpointConfig.ExternalizedCheckpointCleanup>

  
  

• In class org.apache.flink.configuration.ExecutionOptions

  
  

  key	setting	type
  execution.checkpointing.snapshot-compression	ExecutionConfig#useSnapshotCompression	ConfigOption<Boolean>
  execution.buffer-timeout	StreamExecutionEnvironment#bufferTimeout	ConfigOption<Duration>

  
  

Options not configurable from Configuration

• ExecutionConfig
  • ExecutionConfig#executionMode
  • ExecutionConfig#numberOfExecutionRetries
  • ExecutionConfig#codeAnalysisMode
  • ExecutionConfig#executionRetryDelay
  • ExecutionConfig#failTaskOnCheckpointError
  • ExecutionConfig#defaultInputDependencyConstraint
  • ExecutionConfig#registeredTypesWithKryoSerializers
  • ExecutionConfig#registeredTypesWithKryoSerializerClasses
  • ExecutionConfig#defaultKryoSerializers
• CheckpointConfig
  • CheckpointConfig#failOnCheckpointingErrors

Existing configuration options: 

There are at least three places where user can configure execution behavior on a per job case: ExecutionConfig, CheckpointConfig, StreamExecutionEnvironment. The following table lists all configuration parameters that are present in those classes. It also discussed suggested keys that we could use for those options.

ExecutionConfig:

CheckpointConfig:

StreamExecutionEnvironment:

Compatibility, Deprecation, and Migration Plan

...