[FLINK-37440] Fix the bug that parallelism.default do not always adopts 1 as the default value. #26277

beliefer · 2025-03-09T01:54:04Z

What is the purpose of the change

This PR proposes to fix the bug that parallelism.default do not always adopts 1 as the default value.
The config parallelism.default defined as follows.

    public static final ConfigOption<Integer> DEFAULT_PARALLELISM =
            ConfigOptions.key("parallelism.default")
                    .intType()
                    .defaultValue(1)
                    .withDescription("Default parallelism for jobs.");

It means the value of parallelism.default will be 1 if users do not specify it.
But the default is not always 1. I checked all the code and found two special case.

createLocalEnvironment in the StreamExecutionEnvironment at line 2248.

    public static LocalStreamEnvironment createLocalEnvironment(Configuration configuration) {
        if (configuration.getOptional(CoreOptions.DEFAULT_PARALLELISM).isPresent()) {
            return new LocalStreamEnvironment(configuration);
        } else {
            Configuration copyOfConfiguration = new Configuration();
            copyOfConfiguration.addAll(configuration);
            copyOfConfiguration.set(CoreOptions.DEFAULT_PARALLELISM, defaultLocalParallelism);
            return new LocalStreamEnvironment(copyOfConfiguration);
        }
    }

registerEnv in the MiniClusterExtension at line 264.

private void registerEnv(InternalMiniClusterExtension internalMiniClusterExtension) {
        final Configuration configuration =
                internalMiniClusterExtension.getMiniCluster().getConfiguration();

        final int defaultParallelism =
                configuration
                        .getOptional(CoreOptions.DEFAULT_PARALLELISM)
                        .orElse(internalMiniClusterExtension.getNumberSlots());

        TestStreamEnvironment.setAsContext(
                internalMiniClusterExtension.getMiniCluster(), defaultParallelism);
        this.streamExecutionEnvironment =
                new TestStreamEnvironment(
                        internalMiniClusterExtension.getMiniCluster(),
                        internalMiniClusterExtension.getNumberSlots());
    }

I think the DEFAULT_PARALLELISM is a open API, we'd better not change the behavior. So I add extra comments for this config.

Brief change log

Fix the bug that parallelism.default do not always adopts 1 as the default value.

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (docs)

beliefer · 2025-03-09T01:54:39Z

ping @1996fanrui cc @davidradl

flinkbot · 2025-03-09T02:21:11Z

CI report:

5ed887b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

davidradl · 2025-03-09T09:36:13Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

@@ -459,7 +459,10 @@ public static String[] mergeListsToArray(List<String> base, List<String> append)
            ConfigOptions.key("parallelism.default")
                    .intType()
                    .defaultValue(1)
-                    .withDescription("Default parallelism for jobs.");
+                    .withDescription("Default parallelism for jobs. There is an exception here. When creating a"


I am not sure what the new description is saying:

Is the description ever a default of 1? If the default is not always 1 - I suggest we consider removing the default though this would have migration implications we would need to think through.

the phrase "There is an exception here." I would not use the word exception; as this is an overloaded word implying there might be an error of some sort.

I do not think the sentence "the return value of Runtime. getRuntime. availableProcessors()" should be included, it refers to java code, that would not be very understandable for someone truing to configure this.

Please can you verify what cases 1 is the default and when the parallelism should be the number of availableProcessors? It would seem to me that using the available processors for parallelism would make a lot of sense for all cases - if a value has not been supplied. Is this the case?

Yes. The default is not always 1. Because remove the default value is a break change, so I just update the description.
Yeah, the word exception looks weird, how about special case?
I checked all the code and found two special case. I have pasted the two case into the PR's description.

@beliefer when you say special cases, do you mean that there are 3 scenarios , default = 1 , case 1 and case 2. If default=1 never happens - then we could remove it - deprecate it.

default=1 still is the main behavior, remove it will break the API compatibility.
Personally, I want remove the the .defaultValue(1)

beliefer · 2025-03-09T11:22:43Z

flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java

@@ -495,7 +495,6 @@ public boolean isCompatibleWith(@Nonnull Configuration configuration) {
                @Override
                public PipelineExecutor getExecutor(@Nonnull Configuration configuration) {
                    return (pipeline, config, classLoader) -> {
-                        final int parallelism = config.get(CoreOptions.DEFAULT_PARALLELISM);


This line is not used at all.

beliefer · 2025-03-10T01:28:07Z

@flinkbot run azure

1996fanrui · 2025-03-10T02:25:14Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

@@ -459,7 +459,8 @@ public static String[] mergeListsToArray(List<String> base, List<String> append)
            ConfigOptions.key("parallelism.default")
                    .intType()
                    .defaultValue(1)
-                    .withDescription("Default parallelism for jobs.");
+                    .withDescription(


https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=66528&view=logs&j=fc5181b0-e452-5c8f-68de-1097947f6483&s=ae4f8708-9994-57d3-c2d7-b892156e7812&t=995c650b-6573-581c-9ce6-7ad4cc038461&l=31682

From the above link, we could check why CI is failed: Documentation is outdated, please regenerate it according to the instructions in flink-docs/README.md.

After updating the description, the document should be updated together via running mvn package -Dgenerate-config-docs -pl flink-docs -am -nsu -DskipTests -Pskip-webui-build.

Thank you for the reminder.

davidradl · 2025-03-10T09:33:08Z

docs/layouts/shortcodes/generated/core_configuration.html

@@ -74,7 +74,7 @@
            <td><h5>parallelism.default</h5></td>
            <td style="word-wrap: break-word;">1</td>
            <td>Integer</td>
-            <td>Default parallelism for jobs.</td>
+            <td>Default parallelism for jobs. There are two special case. The first case is when creating a StreamExecutioneEnvironment, if the value of parallelism.default is not explicitly specified, the number of processors available to the Java virtual machine will be used as the default parallelism. The second case is when creating mini cluster, the number of slots will be used as the default parallelism</td>


nits:
"There are two special case." -> "There are two cases."
"The first case is when creating a StreamExecutioneEnvironment, if the value of parallelism.default is not explicitly specified, the number of processors available to the Java virtual machine will be used as the default parallelism." ->
In the first case, when creating a StreamExecutioneEnvironment without specifying parallelism.default ,the number of processors available to the Java virtual machine will be used."

" The second case is when creating mini cluster, the number of slots will be used as the default parallelism"
-> I am not sure what you mean by mini cluster. The slots are on a task manager - how does this relate?

It's the total number of slots in mini cluster.

davidradl · 2025-03-10T09:35:15Z

I see docs we should update this test too. around " setParallelism(int parallelism) Set the default parallelism for the job."

davidradl · 2025-03-10T09:39:44Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

@@ -459,7 +459,8 @@ public static String[] mergeListsToArray(List<String> base, List<String> append)
            ConfigOptions.key("parallelism.default")
                    .intType()
                    .defaultValue(1)
-                    .withDescription("Default parallelism for jobs.");
+                    .withDescription(


are we removing .defaultValue(1) , as from your above description - it is never used?

No, It used in most scenes.
Personally, I want remove the the .defaultValue(1) but I concern the API compatibility.
I will do it if more committers suggest to remove it.

+1 to keep the default value as it's using in some scenes.

I would add an existing option[1] example here, it's jobmanager.scheduler. Although, the default value is Default, but the default value will be updated to AdaptiveBatch for flink batch job. The description is:

Determines which scheduler implementation is used to schedule tasks. If this option is not explicitly set, batch jobs will use the 'AdaptiveBatch' scheduler as the default, while streaming jobs will default to the 'Default' scheduler.

I think it's similar with parallelism.default, there are some special cases that don't respect the default value.

[1] https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/config/#jobmanager-scheduler

davidradl · 2025-03-10T09:50:10Z

docs/layouts/shortcodes/generated/sink_configuration.html

+            <td><h5>sink.committer.retries</h5></td>
+            <td style="word-wrap: break-word;">10</td>
+            <td>Integer</td>
+            <td>The number of retries a Flink application attempts for committable operations (such as transactions) on retriable errors, as specified by the sink connector, before Flink fails and potentially restarts.</td>


nits:
I see we have copied:
"committable operations (such as transactions)" I do t think this reads well. An operation can take part in a transaction - but is not a transaction

Flink fails -> don't we mean the Flink application / job fails.

This change is generated by mvn package -Dgenerate-config-docs -pl flink-docs -am -nsu -DskipTests -Pskip-webui-build. I don't know the reason yet.

1996fanrui · 2025-03-13T03:37:25Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

@@ -459,7 +459,8 @@ public static String[] mergeListsToArray(List<String> base, List<String> append)
            ConfigOptions.key("parallelism.default")
                    .intType()
                    .defaultValue(1)
-                    .withDescription("Default parallelism for jobs.");
+                    .withDescription(


I would add an existing option[1] example here, it's jobmanager.scheduler. Although, the default value is Default, but the default value will be updated to AdaptiveBatch for flink batch job. The description is:

Determines which scheduler implementation is used to schedule tasks. If this option is not explicitly set, batch jobs will use the 'AdaptiveBatch' scheduler as the default, while streaming jobs will default to the 'Default' scheduler.

I think it's similar with parallelism.default, there are some special cases that don't respect the default value.

[1] https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/config/#jobmanager-scheduler

1996fanrui · 2025-03-13T03:43:36Z

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

@@ -470,7 +470,8 @@ public static String[] mergeListsToArray(List<String> base, List<String> append)
            ConfigOptions.key("parallelism.default")
                    .intType()
                    .defaultValue(1)
-                    .withDescription("Default parallelism for jobs.");
+                    .withDescription(
+                            "Default parallelism for jobs. There are two special cases. In the first case, when creating a StreamExecutioneEnvironment without specifying parallelism.default, the number of processors available to the Java virtual machine will be used. In the second case, when creating mini cluster without specifying parallelism.default, the total number of slots in mini cluster will be used.");


I have a question about the first case, does it cover all running modes?

For application and session mode, and standalone, yarn or kubernetes type. Doesn't parallelism.default take effect for all of these cases?

No. The first case only found in the local mode.

1996fanrui · 2025-03-13T03:45:26Z

docs/content/docs/dev/datastream/execution/execution_configuration.md

@@ -51,7 +51,7 @@ The following configuration options are available: (the default is bold)
 With the closure cleaner disabled, it might happen that an anonymous user function is referencing the surrounding class, which is usually not Serializable. This will lead to exceptions by the serializer. The settings are:
 `NONE`: disable the closure cleaner completely, `TOP_LEVEL`: clean only the top-level class without recursing into fields, `RECURSIVE`: clean all the fields recursively.

- `getParallelism()` / `setParallelism(int parallelism)` Set the default parallelism for the job.
+- `getParallelism()` / `setParallelism(int parallelism)` Set the default parallelism for the job. There are two special cases. In the first case, when creating a StreamExecutioneEnvironment without specifying parallelism.default, the number of processors available to the Java virtual machine will be used. In the second case, when creating mini cluster without specifying parallelism.default, the total number of slots in mini cluster will be used.


IIUC, if setParallelism(int parallelism) is called, the default parallelism will be changed. I don't understand why mentioning these 2 cases here.

According to the original comment, users understand the parallelism will be 1 if it is not explicitly set. Users confused if they find more parallelism to be used.

…ts 1 as the default value.

beliefer · 2025-03-18T02:11:47Z

@1996fanrui Could you take a review again?

flinkbot added the component=API/Core label Mar 9, 2025

davidradl reviewed Mar 9, 2025

View reviewed changes

beliefer force-pushed the FLINK-37440 branch 2 times, most recently from f73f080 to 25a7216 Compare March 9, 2025 11:21

beliefer commented Mar 9, 2025

View reviewed changes

1996fanrui reviewed Mar 10, 2025

View reviewed changes

1996fanrui self-assigned this Mar 10, 2025

davidradl reviewed Mar 10, 2025

View reviewed changes

beliefer force-pushed the FLINK-37440 branch from 2f8d58c to c2752e2 Compare March 10, 2025 12:49

1996fanrui reviewed Mar 13, 2025

View reviewed changes

[FLINK-37440] Fix the bug that parallelism.default do not always adop…

5ed887b

…ts 1 as the default value.

beliefer force-pushed the FLINK-37440 branch from bd23052 to 5ed887b Compare March 14, 2025 01:03

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-37440] Fix the bug that parallelism.default do not always adopts 1 as the default value. #26277

[FLINK-37440] Fix the bug that parallelism.default do not always adopts 1 as the default value. #26277

beliefer commented Mar 9, 2025 •

edited

Loading

beliefer commented Mar 9, 2025

flinkbot commented Mar 9, 2025 •

edited

Loading

davidradl Mar 9, 2025 •

edited

Loading

beliefer Mar 9, 2025 •

edited

Loading

davidradl Mar 10, 2025 •

edited

Loading

beliefer Mar 11, 2025

beliefer Mar 9, 2025

beliefer commented Mar 10, 2025

1996fanrui Mar 10, 2025

beliefer Mar 10, 2025

davidradl Mar 10, 2025

beliefer Mar 10, 2025

davidradl commented Mar 10, 2025

davidradl Mar 10, 2025

beliefer Mar 10, 2025

1996fanrui Mar 11, 2025

1996fanrui Mar 13, 2025 •

edited

Loading

beliefer Mar 13, 2025

davidradl Mar 10, 2025

beliefer Mar 10, 2025

1996fanrui Mar 13, 2025 •

edited

Loading

1996fanrui Mar 13, 2025

beliefer Mar 13, 2025

1996fanrui Mar 13, 2025

beliefer Mar 13, 2025

beliefer commented Mar 18, 2025

[FLINK-37440] Fix the bug that parallelism.default do not always adopts 1 as the default value. #26277

Are you sure you want to change the base?

[FLINK-37440] Fix the bug that parallelism.default do not always adopts 1 as the default value. #26277

Conversation

beliefer commented Mar 9, 2025 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

beliefer commented Mar 9, 2025

flinkbot commented Mar 9, 2025 • edited Loading

CI report:

davidradl Mar 9, 2025 • edited Loading

Choose a reason for hiding this comment

beliefer Mar 9, 2025 • edited Loading

Choose a reason for hiding this comment

davidradl Mar 10, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

beliefer commented Mar 10, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidradl commented Mar 10, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

1996fanrui Mar 13, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

1996fanrui Mar 13, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

beliefer commented Mar 18, 2025

beliefer commented Mar 9, 2025 •

edited

Loading

flinkbot commented Mar 9, 2025 •

edited

Loading

davidradl Mar 9, 2025 •

edited

Loading

beliefer Mar 9, 2025 •

edited

Loading

davidradl Mar 10, 2025 •

edited

Loading

1996fanrui Mar 13, 2025 •

edited

Loading

1996fanrui Mar 13, 2025 •

edited

Loading