[FLINK-34566] Pass a FixedThreadPool to set reconciliation parallelism correctly

Author: fengfei02

Diff for: flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java

@@ -137,7 +137,7 @@ private void overrideOperatorConfigs(ConfigurationServiceOverrider overrider) {
             overrider.withExecutorService(Executors.newCachedThreadPool());
         } else {
             LOG.info("Configuring operator with {} reconciliation threads.", parallelism);
-            overrider.withConcurrentReconciliationThreads(parallelism);
+            overrider.withExecutorService(Executors.newFixedThreadPool(parallelism));
         }
 
         if (operatorConf.isJosdkMetricsEnabled()) {

Diff for: flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java

@@ -70,11 +70,22 @@ public void testConfigurationPassedToJOSDK() {
 
         var configService = testOperator.getOperator().getConfigurationService();
 
-        // Test parallelism being passed
+        // Test parallelism being passed expectedly
         var executorService = configService.getExecutorService();
         Assertions.assertInstanceOf(ThreadPoolExecutor.class, executorService);
         ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executorService;
+        for (int i = 0; i < testParallelism * 2; i++) {
+            threadPoolExecutor.execute(
+                    () -> {
+                        try {
+                            Thread.sleep(1000);
+                        } catch (InterruptedException e) {
+                            e.printStackTrace();
+                        }
+                    });
+        }
         Assertions.assertEquals(threadPoolExecutor.getMaximumPoolSize(), testParallelism);
+        Assertions.assertEquals(threadPoolExecutor.getPoolSize(), testParallelism);
 
         // Test label selector being passed
         // We have a label selector for each controller