[FLINK-35116] Bump operator sdk version to 4.8.3

Author: mbalassi

docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html

@@ -304,7 +304,7 @@
         </tr>
         <tr>
             <td><h5>kubernetes.operator.reconcile.parallelism</h5></td>
-            <td style="word-wrap: break-word;">200</td>
+            <td style="word-wrap: break-word;">50</td>
             <td>Integer</td>
             <td>The maximum number of threads running the reconciliation loop. Use -1 for infinite.</td>
         </tr>

docs/layouts/shortcodes/generated/system_section.html

@@ -100,7 +100,7 @@
         </tr>
         <tr>
             <td><h5>kubernetes.operator.reconcile.parallelism</h5></td>
-            <td style="word-wrap: break-word;">200</td>
+            <td style="word-wrap: break-word;">50</td>
             <td>Integer</td>
             <td>The maximum number of threads running the reconciliation loop. Use -1 for infinite.</td>
         </tr>

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java

@@ -93,7 +93,11 @@ protected static void checkObjectCompatibility(
                     // This field was removed from Kubernetes ObjectMeta v1 in 1.25 as it was unused
                     // for a long time. If set for any reason (very unlikely as it does nothing),
                     // the property will be dropped / ignored by the api server.
-                    if (!fieldPath.endsWith(".metadata.clusterName")) {
+                    if (!fieldPath.endsWith(".metadata.clusterName")
+                            // This claims field was removed in Kubernetes 1.28 as it was mistakenly
+                            // added in the first place. For more context please refer to
+                            // https://github.com/kubernetes/api/commit/8b14183
+                            && !fieldPath.contains(".volumeClaimTemplate.spec.resources.claims")) {
                         err(fieldPath + " has been removed");
                     }
                 } else {

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.withExecutorService(Executors.newFixedThreadPool(parallelism));
+            overrider.withConcurrentReconciliationThreads(parallelism);
         }
 
         if (operatorConf.isJosdkMetricsEnabled()) {

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

@@ -266,8 +266,7 @@ private static LeaderElectionConfiguration getLeaderElectionConfig(Configuration
                 null,
                 conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_LEASE_DURATION),
                 conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RENEW_DEADLINE),
-                conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RETRY_PERIOD),
-                null);
+                conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RETRY_PERIOD));
     }
 
     private static Optional<String> getEnv(String key) {

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java

@@ -205,9 +205,9 @@ private KubernetesResourceNamespaceMetricGroup getResourceNsMg(
 
         Class<? extends AbstractFlinkResource<?, ?>> resourceClass;
 
-        if (resourceGvk.kind.equals(FlinkDeployment.class.getSimpleName())) {
+        if (resourceGvk.getKind().equals(FlinkDeployment.class.getSimpleName())) {
             resourceClass = FlinkDeployment.class;
-        } else if (resourceGvk.kind.equals(FlinkSessionJob.class.getSimpleName())) {
+        } else if (resourceGvk.getKind().equals(FlinkSessionJob.class.getSimpleName())) {
             resourceClass = FlinkSessionJob.class;
         } else {
             return Optional.empty();

flink-kubernetes-operator/src/main/resources/META-INF/NOTICE

@@ -47,10 +47,10 @@ This project bundles the following dependencies under the Apache Software Licens
 - io.fabric8:kubernetes-model-scheduling:jar:6.8.1
 - io.fabric8:kubernetes-model-storageclass:jar:6.8.1
 - io.fabric8:zjsonpatch:jar:0.3.0
-- io.javaoperatorsdk:operator-framework-core:jar:4.4.4
-- io.javaoperatorsdk:operator-framework:jar:4.4.4
+- io.javaoperatorsdk:operator-framework-core:jar:4.8.3
+- io.javaoperatorsdk:operator-framework:jar:4.8.3
 - org.apache.commons:commons-compress:1.21
-- org.apache.commons:commons-lang3:3.13.0
+- org.apache.commons:commons-lang3:3.14.0
 - org.apache.commons:commons-math3:3.6.1
 - org.apache.commons:commons-text:jar:1.10.0
 - org.apache.logging.log4j:log4j-1.2-api:2.17.1

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

@@ -70,21 +70,10 @@ public void testConfigurationPassedToJOSDK() {
 
         var configService = testOperator.getOperator().getConfigurationService();
 
-        // Test parallelism being passed expectedly
+        // Test parallelism being passed
         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.getPoolSize(), testParallelism);
         Assertions.assertEquals(threadPoolExecutor.getMaximumPoolSize(), testParallelism);
 
         // Test label selector being passed

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java

@@ -51,6 +51,7 @@
 import io.fabric8.mockwebserver.utils.ResponseProvider;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.IndexedResourceCache;
 import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
 import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
@@ -497,5 +498,10 @@ public KubernetesClient getClient() {
         public ExecutorService getWorkflowExecutorService() {
             throw new UnsupportedOperationException("Not implemented");
         }
+
+        @Override
+        public IndexedResourceCache<T> getPrimaryCache() {
+            return null;
+        }
     }
 }

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java

@@ -1013,7 +1013,7 @@ public void testBlockingDeploymentDeletion() {
                         namespace, deploymentName);
         String watchUrl =
                 String.format(
-                        "/apis/apps/v1/namespaces/%s/deployments?fieldSelector=metadata.name%%3D%s&timeoutSeconds=600&allowWatchBookmarks=true&watch=true",
+                        "/apis/apps/v1/namespaces/%s/deployments?allowWatchBookmarks=true&fieldSelector=metadata.name%%3D%s&timeoutSeconds=600&watch=true",
                         namespace, deploymentName);
 
         var flinkService = new TestingService(null);