Status

Current state: Implemented

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

JIRA: Unable to render Jira issues macro, execution error.

Released: 

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

Motivation

Currently the runtime parameters can only be configured via setters in ExecutionConfig / CheckpointConfig / StreamExecutionEnvironment via Java or Scala code. This has some drawbacks:

• User must have access to the StreamExecutionEnvironment. We face this problem in the unified TableEnvironmen mentioned in FLIP-32. There, users do not see the underlying StreamExecutionEnvironment.
• It’s impossible to set the parameters from command line. We face this issue e.g. in SQL Client.
• It’s impossible to have job specific default values for execution parameters provided from a file.
• Moreover some of the config options do not have a string serializable options, which makes it even harder to implement the two previous items.
• These options do not have descriptions in documentation.

We suggest to improve the runtime configuration so that it’s configurable from a Configuration object and integrates nicely in a possible hierarchy of configuration layers in the future.

Assumption: 

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(ReadableConfig configuration, ClassLoader classLoader)
• void CheckpointConfig.configure(ReadableConfig)
• void StreamExecutionEnvironment.configure(ReadableConfig 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("pipeline.closure-cleaner-level")

            .enumType(ClosureCleanerLevel.class)

            .defaultValue(ClosureCleanerLevel.RECURSIVE);


    void configure(ConfigurationReader conf) {

             ...

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

                 this.closureCleanerLevel = level;

            })

            ...

    }

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

The changes are backward compatible. It introduces an additional way of configuring ExecutionConfig, CheckpointConfig & StreamExecutionEnvironment. It reuses existing options wherever possible.

Test Plan

Covered by unit tests

Rejected Alternatives

---