diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index d068062..918b407 100644
--- a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
+++ b/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>
diff --git a/docs/layouts/shortcodes/generated/system_section.html b/docs/layouts/shortcodes/generated/system_section.html
index aa053c2..ec30301 100644
--- a/docs/layouts/shortcodes/generated/system_section.html
+++ b/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>
diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java
index 658e25a..faf6024 100644
--- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java
+++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java
@@ -93,7 +93,11 @@
                     // 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 {
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
index a5846f5..0ecd7c8 100644
--- 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 @@
             overrider.withExecutorService(Executors.newCachedThreadPool());
         } else {
             LOG.info("Configuring operator with {} reconciliation threads.", parallelism);
-            overrider.withExecutorService(Executors.newFixedThreadPool(parallelism));
+            overrider.withConcurrentReconciliationThreads(parallelism);
         }
 
         if (operatorConf.isJosdkMetricsEnabled()) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
index 9fcdb6a..225bebd 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
@@ -266,8 +266,7 @@
                 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) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
index 679aa57..47165f7 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
@@ -205,9 +205,9 @@
 
         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();
diff --git a/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE b/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
index 1b7d232..bd46972 100644
--- a/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
+++ b/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
@@ -47,10 +47,10 @@
 - 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
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
index 6455096..b60551a 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
@@ -70,21 +70,10 @@
 
         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
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
index 2b2e23a..62e9498 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
+++ b/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 ExecutorService getWorkflowExecutorService() {
             throw new UnsupportedOperationException("Not implemented");
         }
+
+        @Override
+        public IndexedResourceCache<T> getPrimaryCache() {
+            return null;
+        }
     }
 }
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index 8e9683d..c4f24ee 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -1013,7 +1013,7 @@
                         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);
diff --git a/pom.xml b/pom.xml
index 823dd03..a5a698d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -75,7 +75,7 @@
         <maven-javadoc-plugin.version>3.3.2</maven-javadoc-plugin.version>
         <git-commit-id-maven-plugin.version>5.0.0</git-commit-id-maven-plugin.version>
 
-        <operator.sdk.version>4.4.4</operator.sdk.version>
+        <operator.sdk.version>4.8.3</operator.sdk.version>
         <operator.sdk.webhook-framework.version>1.1.1</operator.sdk.webhook-framework.version>
 
         <fabric8.version>6.8.1</fabric8.version>