[Feature] Supported apache flink 1.20

.github/workflows/e2e.yml

@@ -136,13 +136,13 @@ jobs:
             class: org.apache.streampark.e2e.cases.VariableManagementTest
 
           - name: FlinkOnRemoteClusterDeployTest
-            class: org.apache.streampark.e2e.cases.Flink118OnRemoteClusterDeployTest
+            class: org.apache.streampark.e2e.cases.Flink120OnRemoteClusterDeployTest
 
           - name: FlinkOnYarnClusterDeployTest
-            class: org.apache.streampark.e2e.cases.Flink118OnYarnClusterDeployTest
+            class: org.apache.streampark.e2e.cases.Flink120OnYarnClusterDeployTest
 
           - name: FlinkSQLOnYarnTest
-            class: org.apache.streampark.e2e.cases.FlinkSQL118OnYarnTest
+            class: org.apache.streampark.e2e.cases.FlinkSQL120OnYarnTest
 
           #- name: Flink117OnRemoteClusterDeployTest
           #  class: org.apache.streampark.e2e.cases.Flink117OnRemoteClusterDeployTest

pom.xml

@@ -93,7 +93,7 @@
         <scala.binary.flink.version>_${scala.binary.version}</scala.binary.flink.version>
         <flink.connector.version>3.2.0-1.18</flink.connector.version>
         <flink.elasticserch.connector.version>3.0.1-1.17</flink.elasticserch.connector.version>
-        <flink.version>1.18.1</flink.version>
+        <flink.version>1.20.1</flink.version>
         <flink.shaded.version>1.8.1</flink.shaded.version>
         <streampark.shaded.version>1.0.0</streampark.shaded.version>
         <streampark.flink.shims.version>1.14</streampark.flink.shims.version>

streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala

@@ -126,7 +126,7 @@ class FlinkVersion(val flinkHome: String) extends Serializable with Logger {
 
   def checkVersion(throwException: Boolean = true): Boolean = {
     version.split("\\.").map(_.trim.toInt) match {
-      case Array(1, v, _) if v >= 12 && v <= 19 => true
+      case Array(1, v, _) if v >= 12 && v <= 20 => true
       case _ =>
         if (throwException) {
           throw new UnsupportedOperationException(s"Unsupported flink version: $version")

streampark-console/streampark-console-service/pom.xml

@@ -586,6 +586,13 @@
                             <version>${project.version}</version>
                             <outputDirectory>${project.build.directory}/shims</outputDirectory>
                         </dependency>
+                        <!-- flink 1.20 support-->
+                        <dependency>
+                            <groupId>org.apache.streampark</groupId>
+                            <artifactId>streampark-flink-shims_flink-1.20_${scala.binary.version}</artifactId>
+                            <version>${project.version}</version>
+                            <outputDirectory>${project.build.directory}/shims</outputDirectory>
+                        </dependency>
                         <!-- flink-submit-core -->
                         <dependency>
                             <groupId>org.apache.streampark</groupId>

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java

@@ -71,7 +71,7 @@ public class EnvInitializer implements ApplicationRunner {
     private final FileFilter fileFilter = p -> !".gitkeep".equals(p.getName());
 
     private static final Pattern PATTERN_FLINK_SHIMS_JAR = Pattern.compile(
-        "^streampark-flink-shims_flink-(1.1[2-9])_(2.12)-(.*).jar$",
+        "^streampark-flink-shims_flink-(1.1[2-9]|1\\.2[0-9])_(2.12)-(.*).jar$",
         Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
 
     @SneakyThrows

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

@@ -484,21 +484,21 @@ public String checkSavepointPath(FlinkApplication appParam) throws Exception {
             final String pathPart = uri.getPath();
             String error = null;
             if (scheme == null) {
-                error = "This state.savepoints.dir value "
+                error = "This state-savepoints-dir value "
                     + savepointPath
                     + " scheme (hdfs://, file://, etc) of  is null. Please specify the file system scheme explicitly in the URI.";
             } else if (pathPart == null) {
-                error = "This state.savepoints.dir value "
+                error = "This state-savepoints-dir value "
                     + savepointPath
                     + " path part to store the checkpoint data in is null. Please specify a directory path for the checkpoint data.";
             } else if (pathPart.isEmpty() || "/".equals(pathPart)) {
-                error = "This state.savepoints.dir value "
+                error = "This state-savepoints-dir value "
                     + savepointPath
                     + " Cannot use the root directory for checkpoints.";
             }
             return error;
         } else {
-            return "When custom savepoint is not set, state.savepoints.dir needs to be set in properties or flink-conf.yaml of application";
+            return "When custom savepoint is not set, state-savepoints-dir needs to be set in properties or flink config of application";
         }
     }
 }

streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkSavepointServiceTest.java

@@ -82,8 +82,8 @@ void cleanTestRecordsInDatabase() {
      */
     @Test
     void testGetSavepointFromDynamicProps() {
-        String propsWithEmptyTargetValue = "-Dstate.savepoints.dir=";
-        String props = "-Dstate.savepoints.dir=hdfs:///test";
+        String propsWithEmptyTargetValue = "-Dexecution.checkpointing.savepoint-dir=";
+        String props = "-Dexecution.checkpointing.savepoint-dir=hdfs:///test";
         FlinkSavepointServiceImpl savepointServiceImpl = (FlinkSavepointServiceImpl) savepointService;
 
         assertThat(savepointServiceImpl.getSavepointFromDynamicProps(null)).isNull();
@@ -114,11 +114,13 @@ void testGetSavepointFromAppCfgIfStreamParkOrSQLJob() {
         app.setJobType(FlinkJobType.FLINK_JAR.getMode());
         assertThat(savepointServiceImpl.getSavepointFromConfig(app)).isNull();
 
+        String ckDir = SAVEPOINT_DIRECTORY.key() + "=hdfs:///test";
+
         // Test for (StreamPark job Or FlinkSQL job) with application config just disabled checkpoint.
         FlinkApplicationConfig appCfg = new FlinkApplicationConfig();
         appCfg.setId(appCfgId);
         appCfg.setAppId(appId);
-        appCfg.setContent("state.savepoints.dir=hdfs:///test");
+        appCfg.setContent(ckDir);
         appCfg.setFormat(ConfigFileTypeEnum.PROPERTIES.getValue());
         configService.save(appCfg);
         assertThat(savepointServiceImpl.getSavepointFromConfig(app)).isNull();
@@ -134,9 +136,10 @@ void testGetSavepointFromAppCfgIfStreamParkOrSQLJob() {
         // Test for configured CHECKPOINTING_INTERVAL
         appCfg.setContent(
             DeflaterUtils.zipString(
-                "state.savepoints.dir=hdfs:///test\n"
+                ckDir + "\n"
                     + String.format("%s=%s", CHECKPOINTING_INTERVAL.key(),
                         "3min")));
+
         configService.updateById(appCfg);
         FlinkEffective effective = new FlinkEffective();
         effective.setTargetId(appCfg.getId());

streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink120OnRemoteClusterDeployTest.java

@@ -0,0 +1,104 @@
+/*
+ * 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.e2e.cases;
+
+import org.apache.streampark.e2e.core.StreamPark;
+import org.apache.streampark.e2e.pages.LoginPage;
+import org.apache.streampark.e2e.pages.common.Constants;
+import org.apache.streampark.e2e.pages.flink.ApacheFlinkPage;
+import org.apache.streampark.e2e.pages.flink.FlinkHomePage;
+import org.apache.streampark.e2e.pages.flink.clusters.ClusterDetailForm;
+import org.apache.streampark.e2e.pages.flink.clusters.FlinkClustersPage;
+import org.apache.streampark.e2e.pages.flink.clusters.RemoteForm;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.openqa.selenium.WebElement;
+import org.openqa.selenium.remote.RemoteWebDriver;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@StreamPark(composeFiles = "docker/flink-1.20-on-remote/docker-compose.yaml")
+public class Flink120OnRemoteClusterDeployTest {
+
+    public static RemoteWebDriver browser;
+
+    private static final String flinkName = "flink-1.20.1";
+
+    private static final String flinkHome = "/opt/flink/";
+
+    private static final String flinkDescription = "description test";
+
+    private static final String flinkClusterName = "flink_1.20.1_cluster_e2e";
+
+    private static final String flinkJobManagerUrl = "http://jobmanager:8081";
+
+    private static final ClusterDetailForm.DeployMode deployMode = ClusterDetailForm.DeployMode.STANDALONE;
+
+    @BeforeAll
+    public static void setUp() {
+        FlinkHomePage flinkHomePage = new LoginPage(browser)
+            .login()
+            .goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkHomePage.class);
+
+        flinkHomePage.createFlinkHome(flinkName, flinkHome, flinkDescription);
+
+        flinkHomePage.goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkClustersPage.class);
+    }
+
+    @Test
+    @Order(1)
+    public void testCreateFlinkCluster() {
+        FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.createFlinkCluster()
+            .<RemoteForm>addCluster(deployMode)
+            .jobManagerURL(flinkJobManagerUrl)
+            .clusterName(flinkClusterName)
+            .flinkVersion(flinkName)
+            .submit();
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> assertThat(flinkClustersPage.flinkClusterList)
+                    .as("Flink clusters list should contain newly-created application")
+                    .extracting(WebElement::getText)
+                    .anyMatch(it -> it.contains(flinkClusterName)));
+    }
+
+    @Test
+    @Order(5)
+    public void testDeleteFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.deleteFlinkCluster(flinkClusterName);
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> {
+                    browser.navigate().refresh();
+                    Thread.sleep(Constants.DEFAULT_SLEEP_MILLISECONDS);
+                    assertThat(flinkClustersPage.flinkClusterList)
+                        .noneMatch(it -> it.getText().contains(flinkClusterName));
+                });
+    }
+}

streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink120OnYarnClusterDeployTest.java

@@ -0,0 +1,152 @@
+/*
+ * 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.e2e.cases;
+
+import org.apache.streampark.e2e.core.StreamPark;
+import org.apache.streampark.e2e.pages.LoginPage;
+import org.apache.streampark.e2e.pages.common.Constants;
+import org.apache.streampark.e2e.pages.flink.ApacheFlinkPage;
+import org.apache.streampark.e2e.pages.flink.FlinkHomePage;
+import org.apache.streampark.e2e.pages.flink.clusters.ClusterDetailForm;
+import org.apache.streampark.e2e.pages.flink.clusters.FlinkClustersPage;
+import org.apache.streampark.e2e.pages.flink.clusters.YarnSessionForm;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.openqa.selenium.WebElement;
+import org.openqa.selenium.remote.RemoteWebDriver;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@StreamPark(composeFiles = "docker/flink-1.20-on-yarn/docker-compose.yaml")
+public class Flink120OnYarnClusterDeployTest {
+
+    public static RemoteWebDriver browser;
+
+    private static final String flinkName = "flink-1.20.1";
+
+    private static final String flinkHome = "/flink-1.20.1";
+
+    private static final String flinkDescription = "description test";
+
+    private static final String flinkClusterName = "flink_1.20.1_cluster_e2e";
+
+    private static final String flinkClusterNameEdited = "flink_1.20.1_cluster_e2e_edited";
+
+    private static final ClusterDetailForm.DeployMode deployMode = ClusterDetailForm.DeployMode.YARN_SESSION;
+
+    @BeforeAll
+    public static void setup() {
+        FlinkHomePage flinkHomePage = new LoginPage(browser)
+            .login()
+            .goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkHomePage.class);
+
+        flinkHomePage.createFlinkHome(flinkName, flinkHome, flinkDescription);
+
+        flinkHomePage.goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkClustersPage.class);
+    }
+
+    @Test
+    @Order(1)
+    public void testCreateFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.createFlinkCluster()
+            .<YarnSessionForm>addCluster(deployMode)
+            .resolveOrder(YarnSessionForm.ResolveOrder.CHILD_FIRST)
+            .clusterName(flinkClusterName)
+            .flinkVersion(flinkName)
+            .submit();
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> assertThat(flinkClustersPage.flinkClusterList)
+                    .as("Flink clusters list should contain newly-created application")
+                    .extracting(WebElement::getText)
+                    .anyMatch(it -> it.contains(flinkClusterName)));
+    }
+
+    @Test
+    @Order(2)
+    public void testEditFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.editFlinkCluster(flinkClusterName)
+            .<YarnSessionForm>addCluster(deployMode)
+            .clusterName(flinkClusterNameEdited)
+            .submit();
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> assertThat(flinkClustersPage.flinkClusterList)
+                    .as("Flink clusters list should contain edited application")
+                    .extracting(WebElement::getText)
+                    .anyMatch(it -> it.contains(flinkClusterNameEdited)));
+    }
+
+    @Test
+    @Order(3)
+    public void testStartFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.startFlinkCluster(flinkClusterNameEdited);
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> assertThat(flinkClustersPage.flinkClusterList)
+                    .as("Flink clusters list should contain running application")
+                    .extracting(WebElement::getText)
+                    .anyMatch(it -> it.contains("RUNNING")));
+    }
+
+    @Test
+    @Order(4)
+    public void testStopFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.stopFlinkCluster(flinkClusterNameEdited);
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> assertThat(flinkClustersPage.flinkClusterList)
+                    .as("Flink clusters list should contain canceled application")
+                    .extracting(WebElement::getText)
+                    .anyMatch(it -> it.contains("CANCELED")));
+    }
+
+    @Test
+    @Order(5)
+    public void testDeleteFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.deleteFlinkCluster(flinkClusterNameEdited);
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> {
+                    browser.navigate().refresh();
+                    Thread.sleep(Constants.DEFAULT_SLEEP_MILLISECONDS);
+                    assertThat(flinkClustersPage.flinkClusterList)
+                        .noneMatch(it -> it.getText().contains(flinkClusterNameEdited));
+                });
+    }
+}