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

Author: beliefer

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 LocalStreamEnvironment 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.
 
 - `getMaxParallelism()` / `setMaxParallelism(int parallelism)` Set the default maximum parallelism for the job. This setting determines the maximum degree of parallelism and specifies the upper limit for dynamic scaling.
 

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 cases. In the first case, when creating a LocalStreamEnvironment 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.</td>
         </tr>
     </tbody>
 </table>

docs/layouts/shortcodes/generated/sink_configuration.html

@@ -0,0 +1,18 @@
+<table class="configuration table table-bordered">
+    <thead>
+        <tr>
+            <th class="text-left" style="width: 20%">Key</th>
+            <th class="text-left" style="width: 15%">Default</th>
+            <th class="text-left" style="width: 10%">Type</th>
+            <th class="text-left" style="width: 55%">Description</th>
+        </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <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>
+        </tr>
+    </tbody>
+</table>

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);
                         final JobGraph jobGraph = streamGraph.getJobGraph();
                         // The job graphs from different cases are generated from the same stream
                         // graph, resulting in the same job ID, which can lead to exceptions.

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 LocalStreamEnvironment 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.");
 
     // ------------------------------------------------------------------------
     //  file systems