Leonard Xu created FLINK-14861:
----------------------------------
Summary: parallelism.default in flink-conf.yaml do not work which is a bug imported by[FLINK-14745]
Key: FLINK-14861
URL:
https://issues.apache.org/jira/browse/FLINK-14861 Project: Flink
Issue Type: Bug
Components: Client / Job Submission
Affects Versions: 1.10.0
Reporter: Leonard Xu
Fix For: 1.10.0
I set parameter "parallelism.default" in flink-conf.yaml, but it's do not work any more when I rebased my branch to master. I debug and find it's a bug imported by [FLINK-14745](
https://issues.apache.org/jira/browse/FLINK-14745).
Detail:
{code:java}
// ExecutionConfigAccessor#fromProgramOptions
public static ExecutionConfigAccessor fromProgramOptions(final ProgramOptions options, final List<URL> jobJars) {
checkNotNull(options);
checkNotNull(jobJars);
final Configuration configuration = new Configuration();
if (options.getParallelism() != ExecutionConfig.PARALLELISM_DEFAULT) {
configuration.setInteger(CoreOptions.DEFAULT_PARALLELISM, options.getParallelism());
}
configuration.setBoolean(DeploymentOptions.ATTACHED, !options.getDetachedMode());
configuration.setBoolean(DeploymentOptions.SHUTDOWN_IF_ATTACHED, options.isShutdownOnAttachedExit());
ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.CLASSPATHS, options.getClasspaths(), URL::toString);
ConfigUtils.encodeCollectionToConfig(configuration, PipelineOptions.JARS, jobJars, URL::toString);
SavepointRestoreSettings.toConfiguration(options.getSavepointRestoreSettings(), configuration);
return new ExecutionConfigAccessor(configuration);
}{code}
[1]. function executionConfigAccessor.getParallelism() will return 1 rather than -1 when options.getParallelism() == ExecutionConfig.PARALLELISM_DEFAULT because
when getParallelism() function will return the defaultValue of CoreOptions.DEFAULT_PARALLELISM.
{code:java}
// ExecutionConfigAccessor.java
public int getParallelism() {
return configuration.getInteger(CoreOptions.DEFAULT_PARALLELISM);
}
// Configuration.java
public int getInteger(ConfigOption<Integer> configOption) {
return getOptional(configOption)
.orElseGet(configOption::defaultValue);
}{code}
And function executionConfigAccessor.getParallelism() still return 1 when options.getParallelism() == 1.
So， the following code in CliFrontend.java will never reach if user not set parallelism in flink run command line.
{code:java}
// CliFrontend.java
int parallelism = executionParameters.getParallelism() == -1 ? defaultParallelism : executionParameters.getParallelism();{code}
[2]and another position, I think we should keep two line which deleted in FLINK-14745--.
{code:java}
//
int userParallelism = executionParameters.getParallelism();
LOG.debug("User parallelism is set to {}", userParallelism);
{code}
*if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) \{
userParallelism = defaultParallelism;
}*
{code:java}
executeProgram(program, client, userParallelism, executionParameters.getDetachedMode());
{code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
Locked
