[K8s] Migrate streampark-flink-kubernetes from Scala to Java (#4465)

* [Migrate] #4451 Migrate streampark-flink-kubernetes from Scala to Java

Convert K8s integration module to Java with Jackson REST DTOs replacing
json4s, preserve watcher/cache behavior, and update console enum bridges.

* [CI] Fix Sonar reliability issues for #4465 k8s migration

Handle InterruptedException in watcher CompletableFuture joins, remove
redundant Optional null checks, and replace backtracking-prone path regex.

* [CI] Fix Sonar code smell issues for #4465 k8s migration

Address 32 Sonar CODE_SMELL findings: extract string constants (S1192),
refactor nested try blocks (S1141), simplify complex methods (S3776/S135),
use @Builder.Default static constants (S1170), and fix comment/path issues.


* [CI] Fix Scala/Java interop for ClusterKey in session client

Update KubernetesNativeSessionClient to use Java ClusterKey builder and
Optional.orElseThrow after k8s module migration to Java.

* [CI] Harden E2E against flaky Docker Hub ryuk pulls

Use GHCR ryuk mirror with retry pre-pull for testcontainers images to
avoid transient registry-1.docker.io timeouts in CI.


* [CI] Fix E2E ryuk pull by removing invalid GHCR mirror

ghcr.io/testcontainers/ryuk:0.7.0 does not exist (manifest unknown),
which broke all E2E jobs. Keep Docker Hub pre-pull with retry only.

* [CI] Fix Sonar S1170 @Builder.Default issues in k8s models

Use inline literal defaults instead of static constants referenced by
instance fields, which triggered "Make this final field static too".

Author: shangeyao

.github/workflows/e2e.yml

@@ -198,6 +198,22 @@ jobs:
         run: |
           docker load -i /tmp/standalone-image.tar \
           && rm -f /tmp/standalone-image.tar
+      - name: Pre-pull Testcontainers images
+        run: |
+          pull_with_retry() {
+            local image="$1"
+            for attempt in $(seq 1 8); do
+              if docker pull "$image"; then
+                return 0
+              fi
+              echo "Failed to pull $image (attempt $attempt), retrying in ${attempt}0s..."
+              sleep $((attempt * 10))
+            done
+            echo "Failed to pull $image after 8 attempts"
+            return 1
+          }
+          pull_with_retry testcontainers/ryuk:0.7.0
+          pull_with_retry selenium/standalone-chrome:4.13.0 || true
       - name: Run Test
         run: |
           ./mvnw -B -f streampark-e2e/pom.xml -am \

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/FlinkAppStateEnum.java

@@ -21,8 +21,6 @@
 
 import lombok.Getter;
 
-import scala.Enumeration;
-
 /** Describe the status of Flink Application */
 @Getter
 public enum FlinkAppStateEnum {
@@ -149,16 +147,16 @@ public static boolean isLost(Integer appState) {
     @Deprecated
     public static class Bridge {
 
-        /** covert from org.apache.streampark.flink.k8s.enums.FlinkJobState */
-        public static FlinkAppStateEnum fromK8sFlinkJobState(Enumeration.Value flinkJobState) {
-            if (FlinkJobState.K8S_INITIALIZING() == flinkJobState) {
+        /** covert from org.apache.streampark.flink.kubernetes.enums.FlinkJobState */
+        public static FlinkAppStateEnum fromK8sFlinkJobState(FlinkJobState flinkJobState) {
+            if (FlinkJobState.K8S_INITIALIZING == flinkJobState) {
                 return INITIALIZING;
             }
-            return getState(flinkJobState.toString());
+            return getState(flinkJobState.name());
         }
 
-        /** covert to org.apache.streampark.flink.k8s.enums.FlinkJobState */
-        public static Enumeration.Value toK8sFlinkJobState(FlinkAppStateEnum flinkAppStateEnum) {
+        /** covert to org.apache.streampark.flink.kubernetes.enums.FlinkJobState */
+        public static FlinkJobState toK8sFlinkJobState(FlinkAppStateEnum flinkAppStateEnum) {
             return FlinkJobState.of(flinkAppStateEnum.name());
         }
     }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationInfoServiceImpl.java

@@ -370,10 +370,16 @@ public String k8sStartLog(Long id, Integer offset, Integer limit) throws Excepti
             "Job deployMode must be kubernetes-session|kubernetes-application.");
 
         CompletableFuture<String> future = CompletableFuture.supplyAsync(
-            () -> KubernetesDeploymentHelper.watchDeploymentLog(
-                application.getK8sNamespace(),
-                application.getJobName(),
-                application.getJobId()));
+            () -> {
+                try {
+                    return KubernetesDeploymentHelper.watchDeploymentLog(
+                        application.getK8sNamespace(),
+                        application.getJobName(),
+                        application.getJobId());
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            });
 
         return future
             .exceptionally(

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/watcher/FlinkK8sChangeEventListener.java

@@ -47,8 +47,6 @@
 import java.util.Date;
 import java.util.concurrent.Executor;
 
-import scala.Enumeration;
-
 import static org.apache.streampark.console.core.enums.FlinkAppStateEnum.Bridge.fromK8sFlinkJobState;
 import static org.apache.streampark.console.core.enums.FlinkAppStateEnum.Bridge.toK8sFlinkJobState;
 
@@ -161,9 +159,9 @@ public void subscribeCheckpointChange(FlinkJobCheckpointChangeEvent event) {
     }
 
     private void setByJobStatusCV(FlinkApplication app, JobStatusCV jobStatus) {
-        // infer the final flink job state
-        Enumeration.Value state = FlinkJobStatusWatcher.inferFlinkJobStateFromPersist(
-            jobStatus.jobState(), toK8sFlinkJobState(app.getStateEnum()));
+        FlinkJobState state =
+            FlinkJobStatusWatcher.inferFlinkJobStateFromPersist(
+                jobStatus.jobState(), toK8sFlinkJobState(app.getStateEnum()));
 
         // corrective start-time / end-time / duration
         long preStartTime = app.getStartTime() != null ? app.getStartTime().getTime() : 0;

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeSessionClient.scala

@@ -63,19 +63,20 @@ object KubernetesNativeSessionClient extends KubernetesNativeClientTrait with Lo
       fatJar: File): SubmitResponse = {
 
     // get jm rest url of flink session cluster
-    val clusterKey = ClusterKey(
-      FlinkK8sDeployMode.SESSION,
-      submitRequest.kubernetesNamespace,
-      submitRequest.clusterId)
+    val clusterKey = ClusterKey.builder()
+      .executeMode(FlinkK8sDeployMode.SESSION)
+      .namespace(submitRequest.kubernetesNamespace)
+      .clusterId(submitRequest.clusterId)
+      .build()
     val jmRestUrl = KubernetesRetriever
       .retrieveFlinkRestUrl(clusterKey)
-      .getOrElse(
-        throw new Exception(
+      .orElseThrow(() =>
+        new Exception(
           s"[flink-submit] retrieve flink session rest url failed, clusterKey=$clusterKey"))
     // submit job via rest api
     val jobId =
       FlinkSessionSubmitHelper.submitViaRestApi(jmRestUrl, fatJar, flinkConfig)
-    SubmitResponse(clusterKey.clusterId, flinkConfig.toMap, jobId, jmRestUrl)
+    SubmitResponse(clusterKey.clusterId(), flinkConfig.toMap, jobId, jmRestUrl)
   }
 
   /** Submit flink session job with building JobGraph via ClusterClient api. */

streampark-flink/streampark-flink-kubernetes/pom.xml

@@ -31,7 +31,6 @@
     <properties>
         <apache.httpclient5.version>5.1</apache.httpclient5.version>
         <snakeyaml.version>2.0</snakeyaml.version>
-        <scalatest.skiptests>true</scalatest.skiptests>
     </properties>
 
     <dependencies>
@@ -123,9 +122,15 @@
         </dependency>
 
         <dependency>
-            <groupId>org.json4s</groupId>
-            <artifactId>json4s-jackson_${scala.binary.version}</artifactId>
-            <scope>provided</scope>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>${jackson.version}</version>
         </dependency>
 
         <!-- snake yaml -->
@@ -157,6 +162,12 @@
             <artifactId>lombok</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
     <build>

streampark-flink/streampark-flink-kubernetes/src/main/java/org/apache/streampark/flink/kubernetes/CacheKey.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.kubernetes;
+
+import org.apache.streampark.common.util.Utils;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public final class CacheKey implements Serializable {
+
+    private final Long key;
+
+    public CacheKey(Long key) {
+        this.key = key;
+    }
+
+    public Long key() {
+        return key;
+    }
+
+    @Override
+    public int hashCode() {
+        return Utils.hashCode(key);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof CacheKey)) {
+            return false;
+        }
+        CacheKey that = (CacheKey) obj;
+        return Objects.equals(key, that.key);
+    }
+}

streampark-flink/streampark-flink-kubernetes/src/main/java/org/apache/streampark/flink/kubernetes/ChangeEventBus.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.kubernetes;
+
+import org.apache.streampark.common.util.ThreadUtils;
+
+import com.google.common.eventbus.AsyncEventBus;
+import com.google.common.eventbus.EventBus;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class ChangeEventBus {
+
+    private static final int CPU_NUM = Math.max(4, Runtime.getRuntime().availableProcessors() * 2);
+
+    private final ThreadPoolExecutor execPool =
+        new ThreadPoolExecutor(
+            CPU_NUM,
+            CPU_NUM * 5,
+            60L,
+            TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(),
+            ThreadUtils.threadFactory("streampark-k8s-watching-thread"));
+
+    final AsyncEventBus asyncEventBus =
+        new AsyncEventBus("[StreamPark][flink-k8s]AsyncEventBus", execPool);
+
+    final EventBus syncEventBus = new EventBus("[StreamPark][flink-k8s]SyncEventBus");
+
+    public void postAsync(Object event) {
+        asyncEventBus.post(event);
+    }
+
+    public void postSync(Object event) {
+        syncEventBus.post(event);
+    }
+
+    public void registerListener(Object listener) {
+        asyncEventBus.register(listener);
+        syncEventBus.register(listener);
+    }
+}