[FLINK-18019] The configuration specified in TableConfig may not take effect in certain cases

• Type: Sub-task
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.10.0, 1.11.0
• Component/s: Table SQL / Legacy Planner, Table SQL / Planner
• Assignee: Unassigned
• Reporter: Dian Fu
• Created: 29/May/20 03:17
• Updated: 20/Jan/21 12:05

Description

Currently If the following configuration is configured in flink-conf.yaml:

state.backend: rocksdb
state.checkpoints.dir: file:///tmp/flink-checkpoints

and the following configuration is configured via TableConfig:

tableConfig.getConfiguration().setString("state.backend.rocksdb.memory.fixed-per-slot", "200MB")
tableConfig.getConfiguration().setString("taskmanager.memory.task.off-heap.size", "200MB")

Then users submit the job via CliFrontend, the configuration set via TableConfig will not take effect.

Intuitively, it should be that user specified configuration via TableConfig(has higher priority) and the configuration specified via flink-conf.yaml together determines the configuration of a job. However, it doesn't hold in all cases.

The root cause is that only the configuration specified in TableConfig in passed to StreamExecutionEnvironment during translate to plan. For the above case, as state.backend is not specified in TableConfig and so the configuration state.backend.rocksdb.memory.fixed-per-slot will not take effect. Please note that in above example, the state backend actually used will be RocksDB without the configuration state.backend.rocksdb.memory.fixed-per-slot and taskmanager.memory.task.off-heap.size.