[FLINK-34566] Pass a FixedThreadpool to make reconciliation parallelism was set correctly

fengfei02 · fengfei02 · commit 71a3417aab7d · 2024-03-05T15:47:49.000+08:00
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java b/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()) {

Original file line number	Diff line number	Diff line change
`@@ -137,7 +137,7 @@ private void overrideOperatorConfigs(ConfigurationServiceOverrider overrider) {`
`137`	`137`	`overrider.withExecutorService(Executors.newCachedThreadPool());`
`138`	`138`	`} else {`
`139`	`139`	`LOG.info("Configuring operator with {} reconciliation threads.", parallelism);`
`140`		`- overrider.withConcurrentReconciliationThreads(parallelism);`
	`140`	`+ overrider.withExecutorService(Executors.newFixedThreadPool(parallelism));`
`141`	`141`	`}`
`142`	`142`
`143`	`143`	`if (operatorConf.isJosdkMetricsEnabled()) {`