[FLINK-37885] Upgrade to JOSDK 5.1

Author: csviri

.github/workflows/ci.yml

@@ -37,7 +37,7 @@ jobs:
     name: maven build
     strategy:
       matrix:
-        java-version: [ 11, 17, 21 ]
+        java-version: [ 17, 21 ]
     steps:
       - uses: actions/checkout@v4
       - name: Set up JDK ${{ matrix.java-version }}
@@ -76,7 +76,7 @@ jobs:
     strategy:
       matrix:
         http-client: [ "okhttp", "jdk", "jetty", "vertx" ]
-        java-version: [ "11", "17", "21" ]
+        java-version: [ "17", "21" ]
     uses: ./.github/workflows/e2e.yaml
     with:
       java-version: ${{ matrix.java-version }}
@@ -89,7 +89,7 @@ jobs:
     strategy:
       matrix:
         http-client: [ "okhttp" ]
-        java-version: [ "11", "17"]
+        java-version: [ "17"]
         flink-version:
           - "v2_0"
           - "v1_20"
@@ -142,7 +142,7 @@ jobs:
 
     uses: ./.github/workflows/e2e.yaml
     with:
-      java-version: 11
+      java-version: 17
       flink-version: ${{ matrix.flink-version }}
       test: ${{ matrix.test }}
       namespace: "flink"
@@ -211,7 +211,7 @@ jobs:
             test: test_snapshot.sh
     uses: ./.github/workflows/e2e.yaml
     with:
-      java-version: 11
+      java-version: 17
       flink-version: ${{ matrix.flink-version }}
       test: ${{ matrix.test }}
       mode: ${{ matrix.mode }}

.github/workflows/docker_push.yml

@@ -45,7 +45,7 @@ jobs:
           image: tonistiigi/binfmt:qemu-v7.0.0
           platforms: all
 
-      - name: Set up Docker Buildx
+      - name: Set up Docker Build
         uses: docker/setup-buildx-action@v3
 
       - name: Log in to the Container registry

.github/workflows/publish_snapshot.yml

@@ -34,7 +34,7 @@ jobs:
       - name: Set up JDK 11
         uses: actions/setup-java@v4
         with:
-          java-version: '11'
+          java-version: '17'
           distribution: 'temurin'
       - name: Cache local Maven repository
         uses: actions/cache@v4

Dockerfile

@@ -16,7 +16,7 @@
 # limitations under the License.
 ################################################################################
 # Build
-ARG JAVA_VERSION=11
+ARG JAVA_VERSION=17
 FROM maven:3.8.8-eclipse-temurin-${JAVA_VERSION} AS build
 ARG SKIP_TESTS=true
 ARG HTTP_CLIENT=okhttp

docs/content.zh/docs/development/guide.md

@@ -36,7 +36,7 @@ In order to build the operator you need to [clone the git repository]({{< github
 git clone {{< github_repo >}}
 ```
 
-To build from the command line, it is necessary to have **Maven 3** and a **Java Development Kit** (JDK) installed. Please note that Flink Kubernetes Operator requires **Java 11**.
+To build from the command line, it is necessary to have **Maven 3** and a **Java Development Kit** (JDK) installed. Please note that Flink Kubernetes Operator requires **Java 17**.
 
 To build the project, you can use the following command:
 

docs/content/docs/development/guide.md

@@ -36,7 +36,7 @@ In order to build the operator you need to [clone the git repository]({{< github
 git clone {{< github_repo >}}
 ```
 
-To build from the command line, it is necessary to have **Maven 3** and a **Java Development Kit** (JDK) installed. Please note that Flink Kubernetes Operator requires **Java 11**.
+To build from the command line, it is necessary to have **Maven 3** and a **Java Development Kit** (JDK) installed. Please note that Flink Kubernetes Operator requires **Java 17**.
 
 To build the project, you can use the following command:
 

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

@@ -97,7 +97,13 @@ protected static void checkObjectCompatibility(
                             // 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")) {
+                            && !fieldPath.contains(".volumeClaimTemplate.spec.resources.claims")
+                            && !fieldPath.contains(
+                                    ".spec.taskManager.podTemplate.spec.resourceClaims.items.source")
+                            && !fieldPath.contains(
+                                    ".spec.jobManager.podTemplate.spec.resourceClaims.items.source")
+                            && !fieldPath.contains(
+                                    ".spec.podTemplate.spec.resourceClaims.items.source")) {
                         err(fieldPath + " has been removed");
                     }
                 } else {

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

@@ -66,6 +66,7 @@
 
 import javax.annotation.Nullable;
 
+import java.time.Duration;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
@@ -150,14 +151,15 @@ private void overrideOperatorConfigs(ConfigurationServiceOverrider overrider) {
             overrider.withMetrics(new OperatorJosdkMetrics(metricGroup, configManager));
         }
 
-        overrider.withTerminationTimeoutSeconds(
-                (int)
+        overrider.withReconciliationTerminationTimeout(
+                Duration.ofSeconds(
                         conf.get(KubernetesOperatorConfigOptions.OPERATOR_TERMINATION_TIMEOUT)
-                                .toSeconds());
+                                .toSeconds()));
 
         overrider.withStopOnInformerErrorDuringStartup(
                 conf.get(KubernetesOperatorConfigOptions.OPERATOR_STOP_ON_INFORMER_ERROR));
 
+        overrider.withUseSSAToPatchPrimaryResource(false);
         var leaderElectionConf = operatorConf.getLeaderElectionConfiguration();
         if (leaderElectionConf != null) {
             overrider.withLeaderElectionConfiguration(leaderElectionConf);

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

@@ -496,13 +496,13 @@ public static String operatorConfigKey(String key) {
                     .withDescription(
                             "Whether informer errors should stop operator startup. If false, the startup will ignore recoverable errors, caused for example by RBAC issues and will retry periodically.");
 
+    public static final int DEFAULT_TERMINATION_TIMEOUT_SECONDS = 10;
+
     @Documentation.Section(SECTION_ADVANCED)
     public static final ConfigOption<Duration> OPERATOR_TERMINATION_TIMEOUT =
             operatorConfig("termination.timeout")
                     .durationType()
-                    .defaultValue(
-                            Duration.ofSeconds(
-                                    ConfigurationService.DEFAULT_TERMINATION_TIMEOUT_SECONDS))
+                    .defaultValue(Duration.ofSeconds(DEFAULT_TERMINATION_TIMEOUT_SECONDS))
                     .withDescription(
                             "Operator shutdown timeout before reconciliation threads are killed.");
 

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java

@@ -43,10 +43,8 @@
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
-import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
 import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
@@ -55,17 +53,13 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
 /** Controller that runs the main reconcile loop for Flink deployments. */
 @ControllerConfiguration
 public class FlinkDeploymentController
-        implements Reconciler<FlinkDeployment>,
-                ErrorStatusHandler<FlinkDeployment>,
-                EventSourceInitializer<FlinkDeployment>,
-                Cleaner<FlinkDeployment> {
+        implements Reconciler<FlinkDeployment>, Cleaner<FlinkDeployment> {
     private static final Logger LOG = LoggerFactory.getLogger(FlinkDeploymentController.class);
 
     private final Set<FlinkResourceValidator> validators;
@@ -185,9 +179,9 @@ private void triggerErrorEvent(
     }
 
     @Override
-    public Map<String, EventSource> prepareEventSources(
+    public List<EventSource<?, FlinkDeployment>> prepareEventSources(
             EventSourceContext<FlinkDeployment> context) {
-        List<EventSource> eventSources = new ArrayList<>();
+        List<EventSource<?, FlinkDeployment>> eventSources = new ArrayList<>();
         eventSources.add(EventSourceUtils.getSessionJobInformerEventSource(context));
         eventSources.add(EventSourceUtils.getDeploymentInformerEventSource(context));
 
@@ -199,7 +193,7 @@ public Map<String, EventSource> prepareEventSources(
                     "Could not initialize informer for snapshots as the CRD has not been installed!");
         }
 
-        return EventSourceInitializer.nameEventSources(eventSources.toArray(EventSource[]::new));
+        return eventSources;
     }
 
     @Override