[FLINK-38859] Do not apply minimum job vertex parallelism for whole job in native mode

Author: gyfora

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java

@@ -363,9 +363,11 @@ private int getParallelism() {
                     * effectiveConfig.get(TaskManagerOptions.NUM_TASK_SLOTS);
         }
 
-        Optional<Integer> maxOverrideParallelism = getMaxParallelismFromOverrideConfig();
-        if (maxOverrideParallelism.isPresent() && maxOverrideParallelism.get() > 0) {
-            return maxOverrideParallelism.get();
+        if (KubernetesDeploymentMode.STANDALONE.equals(spec.getMode())) {
+            Optional<Integer> maxOverrideParallelism = getMaxParallelismFromOverrideConfig();
+            if (maxOverrideParallelism.isPresent() && maxOverrideParallelism.get() > 0) {
+                return maxOverrideParallelism.get();
+            }
         }
 
         return spec.getJob().getParallelism();

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java

@@ -874,6 +874,38 @@ public void testApplyStandaloneSessionSpec() throws URISyntaxException, IOExcept
                         StandaloneKubernetesConfigOptionsInternal.KUBERNETES_TASKMANAGER_REPLICAS));
     }
 
+    @Test
+    public void testParallelismOverridesOnlyAppliedForStandaloneMode()
+            throws URISyntaxException, IOException {
+        FlinkDeployment dep = ReconciliationUtils.clone(flinkDeployment);
+        dep.getSpec().setTaskManager(new TaskManagerSpec());
+        dep.getSpec().getJob().setParallelism(5);
+        dep.getSpec()
+                .getFlinkConfiguration()
+                .put(PipelineOptions.PARALLELISM_OVERRIDES.key(), "vertex1:10,vertex2:20");
+
+        // Test STANDALONE mode - parallelism overrides should be used
+        dep.getSpec().setMode(KubernetesDeploymentMode.STANDALONE);
+        Configuration configuration =
+                new FlinkConfigBuilder(dep, new Configuration())
+                        .applyFlinkConfiguration()
+                        .applyTaskManagerSpec()
+                        .applyJobOrSessionSpec()
+                        .build();
+        assertEquals(20, configuration.get(CoreOptions.DEFAULT_PARALLELISM));
+
+        // Test NATIVE mode - parallelism overrides should NOT be used, fall back to job
+        // parallelism
+        dep.getSpec().setMode(KubernetesDeploymentMode.NATIVE);
+        configuration =
+                new FlinkConfigBuilder(dep, new Configuration())
+                        .applyFlinkConfiguration()
+                        .applyTaskManagerSpec()
+                        .applyJobOrSessionSpec()
+                        .build();
+        assertEquals(5, configuration.get(CoreOptions.DEFAULT_PARALLELISM));
+    }
+
     @Test
     public void testBuildFrom() throws Exception {
         final Configuration configuration =