ClusterKey code optimization (#2920)

* ClusterKey code optimization * FlinkCheckpointWatcher code optimization
apache · Aug 5, 2023 · b2ec2cd · b2ec2cd
1 parent 2783f0f
commit b2ec2cd
Show file tree

Hide file tree

Showing 5 changed files with 7 additions and 8 deletions.
diff --git a/...k-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/ClusterKey.scala b/...k-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/ClusterKey.scala
@@ -17,12 +17,13 @@
 
 package org.apache.streampark.flink.kubernetes.model
 
+import org.apache.streampark.common.conf.K8sFlinkConfig
 import org.apache.streampark.flink.kubernetes.enums.FlinkK8sExecuteMode
 
 /** flink cluster identifier on kubernetes */
 case class ClusterKey(
     executeMode: FlinkK8sExecuteMode.Value,
-    namespace: String = "default",
+    namespace: String = K8sFlinkConfig.DEFAULT_KUBERNETES_NAMESPACE,
     clusterId: String)
 
 object ClusterKey {

diff --git a/...link-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/TrackId.scala b/...link-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/model/TrackId.scala
@@ -17,6 +17,7 @@
 
 package org.apache.streampark.flink.kubernetes.model
 
+import org.apache.streampark.common.conf.K8sFlinkConfig
 import org.apache.streampark.common.util.Utils
 import org.apache.streampark.flink.kubernetes.enums.FlinkK8sExecuteMode
 
@@ -25,7 +26,7 @@ import scala.util.Try
 /** tracking identifier for flink on kubernetes */
 case class TrackId(
     executeMode: FlinkK8sExecuteMode.Value,
-    namespace: String = "default",
+    namespace: String = K8sFlinkConfig.DEFAULT_KUBERNETES_NAMESPACE,
     clusterId: String,
     appId: Long,
     jobId: String,

diff --git a/...rc/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkCheckpointWatcher.scala b/...rc/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkCheckpointWatcher.scala
@@ -77,8 +77,7 @@ class FlinkCheckpointWatcher(conf: MetricWatcherConfig = MetricWatcherConfig.def
     // get all legal tracking cluster key
     val trackIds: Set[TrackId] = Try(watchController.getActiveWatchingIds())
       .filter(_.nonEmpty)
-      .getOrElse(return
-      )
+      .getOrElse(return None)
     // retrieve flink metrics in thread pool
     val futures: Set[Future[Option[CheckpointCV]]] =
       trackIds.map(

diff --git a/...src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala b/...src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
@@ -99,8 +99,7 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConfig = JobStatusWatcherConfi
       // get all legal tracking ids
       val trackIds = Try(watchController.getAllWatchingIds())
         .filter(_.nonEmpty)
-        .getOrElse(return
-        )
+        .getOrElse(return None)
 
       // retrieve flink job status in thread pool
       val tracksFuture: Set[Future[Option[JobStatusCV]]] = trackIds.map {

diff --git a/...s/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkMetricsWatcher.scala b/...s/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkMetricsWatcher.scala
@@ -77,8 +77,7 @@ class FlinkMetricWatcher(conf: MetricWatcherConfig = MetricWatcherConfig.default
     // get all legal tracking cluster key
     val trackIds: Set[TrackId] = Try(watchController.getActiveWatchingIds())
       .filter(_.nonEmpty)
-      .getOrElse(return
-      )
+      .getOrElse(return None)
     // retrieve flink metrics in thread pool
     val futures: Set[Future[Option[FlinkMetricCV]]] =
       trackIds.map(