[Feat] flink 1.15 support. (#1003)

apache · May 5, 2022 · ea00e39 · ea00e39
1 parent 8386d00
commit ea00e39
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 33 deletions.
diff --git a/scalastyle-config.xml b/scalastyle-config.xml
@@ -208,20 +208,6 @@ This file is divided into 3 sections:
         </customMessage>
     </check>
 
-    <check customId="awaitready" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
-        <parameters>
-            <parameter name="regex">Await\.ready</parameter>
-        </parameters>
-        <customMessage><![CDATA[
-          Are you sure that you want to use Await.ready? In most cases, you should use ThreadUtils.awaitReady instead.
-          If you must use Await.ready, wrap the code block with
-          // scalastyle:off awaitready
-          Await.ready(...)
-          // scalastyle:on awaitready
-        ]]>
-        </customMessage>
-    </check>
-
     <check customId="commons-lang2" level="error" class="org.scalastyle.scalariform.TokenChecker" enabled="true">
         <parameters>
             <parameter name="regex">org\.apache\.commons\.lang\.</parameter>

diff --git a/streamx-console/streamx-console-service/pom.xml b/streamx-console/streamx-console-service/pom.xml
@@ -479,7 +479,6 @@
             </activation>
             <build>
                 <plugins>
-
                     <plugin>
                         <groupId>com.github.eirslett</groupId>
                         <artifactId>frontend-maven-plugin</artifactId>
@@ -548,9 +547,12 @@
         </profile>
 
         <profile>
-            <id>console-flink-shims_default</id>
+            <id>console-flink-shims_2.11</id>
             <activation>
-                <activeByDefault>true</activeByDefault>
+                <property>
+                    <name>scala.binary.version</name>
+                    <value>2.11</value>
+                </property>
             </activation>
             <build>
                 <plugins>
@@ -591,13 +593,6 @@
                                     <version>${project.version}</version>
                                     <outputDirectory>${project.build.directory}/shims</outputDirectory>
                                 </dependency>
-                                <!-- flink 1.15 support-->
-                                <dependency>
-                                    <groupId>com.streamxhub.streamx</groupId>
-                                    <artifactId>streamx-flink-shims_flink-1.15_${scala.binary.version}</artifactId>
-                                    <version>${project.version}</version>
-                                    <outputDirectory>${project.build.directory}/shims</outputDirectory>
-                                </dependency>
                                 <!-- flink-submit-core -->
                                 <dependency>
                                     <groupId>com.streamxhub.streamx</groupId>
@@ -633,11 +628,11 @@
         </profile>
 
         <profile>
-            <id>console-flink-shims_2.11</id>
+            <id>console-flink-shims_2.12</id>
             <activation>
                 <property>
                     <name>scala.binary.version</name>
-                    <value>2.11</value>
+                    <value>2.12</value>
                 </property>
             </activation>
             <build>
@@ -679,6 +674,13 @@
                                     <version>${project.version}</version>
                                     <outputDirectory>${project.build.directory}/shims</outputDirectory>
                                 </dependency>
+                                <!-- flink 1.15 support-->
+                                <dependency>
+                                    <groupId>com.streamxhub.streamx</groupId>
+                                    <artifactId>streamx-flink-shims_flink-1.15_${scala.binary.version}</artifactId>
+                                    <version>${project.version}</version>
+                                    <outputDirectory>${project.build.directory}/shims</outputDirectory>
+                                </dependency>
                                 <!-- flink-submit-core -->
                                 <dependency>
                                     <groupId>com.streamxhub.streamx</groupId>

diff --git a/...rc/main/scala/com/streamxhub/streamx/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala b/...rc/main/scala/com/streamxhub/streamx/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala
@@ -150,11 +150,7 @@ class FlinkJobStatusWatcher(conf: JobStatusWatcherConf = JobStatusWatcherConf.de
       })
     // blocking until all future are completed or timeout is reached
     val allFutureHold = Future.sequence(tracksFuture)
-    Try(
-      // scalastyle:off awaitready
-      Await.ready(allFutureHold, conf.sglTrkTaskTimeoutSec seconds)
-      // scalastyle:on awaitready
-    ).failed.map(_ =>
+    Try(Await.ready(allFutureHold, conf.sglTrkTaskTimeoutSec seconds)).failed.map(_ =>
       logInfo(s"[FlinkJobStatusWatcher] tracking flink job status on kubernetes mode timeout," +
         s" limitSeconds=${conf.sglTrkTaskTimeoutSec}," +
         s" trackingClusterKeys=${trkClusterKeys.mkString(",")}"))

diff --git a/.../src/main/scala/com/streamxhub/streamx/flink/kubernetes/watcher/FlinkMetricsWatcher.scala b/.../src/main/scala/com/streamxhub/streamx/flink/kubernetes/watcher/FlinkMetricsWatcher.scala
@@ -119,10 +119,8 @@ class FlinkMetricWatcher(conf: MetricWatcherConf = MetricWatcherConf.defaultConf
       })
     // blocking until all future are completed or timeout is reached
     Try {
-      // scalastyle:off awaitready
       val futureHold = Future.sequence(trkFutures)
       Await.ready(futureHold, conf.sglTrkTaskTimeoutSec seconds)
-      // scalastyle:on awaitready
     }.failed.map { _ =>
       logError(s"[FlinkMetricWatcher] tracking flink metrics on kubernetes mode timeout," +
         s" limitSeconds=${conf.sglTrkTaskTimeoutSec}," +