[improve](case) Fix spark flink case in jdk17 (#56992)

### What problem does this PR solve?

Fix spark flink case in jdk17

regression-test/suites/connector_p0/spark_connector/spark_connector.groovy

@@ -1,41 +0,0 @@
-// 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.
-
-suite("spark_connector", "connector") {
-
-    def tableName = "spark_connector"
-    sql """DROP TABLE IF EXISTS ${tableName}"""
-    sql new File("""${context.file.parent}/ddl/create.sql""").text
-    logger.info("start delete local spark doris demo jar...")
-    def delete_local_spark_jar = "rm -rf spark-doris-demo.jar".execute()
-    logger.info("start download spark doris demo ...")
-    logger.info("getS3Url ==== ${getS3Url()}")
-    def download_spark_jar = "/usr/bin/curl ${getS3Url()}/regression/spark-doris-connector-demo-jar-with-dependencies.jar --output spark-doris-demo.jar".execute().getText()
-    def out = "/usr/bin/ls -al spark-doris-demo.jar".execute().getText()
-    logger.info("finish download spark doris demo, out: ${out}")
-    def run_cmd = "java -jar spark-doris-demo.jar $context.config.feHttpAddress $context.config.feHttpUser regression_test_connector_p0_spark_connector.$tableName"
-    logger.info("run_cmd : $run_cmd")
-    def proc = run_cmd.execute()
-    def sout = new StringBuilder()
-    def serr = new StringBuilder()
-    proc.consumeProcessOutput(sout, serr)
-    proc.waitForOrKill(1200_000)
-    if (proc.exitValue() != 0) {
-      logger.warn("failed to execute jar: code=${proc.exitValue()}, " + "output: ${sout.toString()}, error: ${serr.toString()}")
-    }
-    qt_select """ select * from $tableName order by order_id"""
-}

regression-test/suites/connector_p0/spark_connector/spark_connector_arrow.groovy

@@ -126,7 +126,7 @@ suite("spark_connector_for_arrow", "connector") {
     sql """DELETE FROM spark_connector_map where id > 0"""
     sql """DELETE FROM spark_connector_struct where id > 0"""
 
-    def jar_name = "spark-doris-connector-3.1_2.12-1.3.0-SNAPSHOT-with-dependencies.jar"
+    def jar_name = "spark-doris-connector-3.4_2.12-1.3.0-SNAPSHOT.jar"
 
     logger.info("start delete local spark doris demo jar...")
     def delete_local_spark_jar = "rm -rf ${jar_name}".execute()
@@ -134,7 +134,18 @@ suite("spark_connector_for_arrow", "connector") {
     logger.info("getS3Url ==== ${getS3Url()}")
     def download_spark_jar = "/usr/bin/curl ${getS3Url()}/regression/${jar_name} --output ${jar_name}".execute().getText()
     logger.info("finish download spark doris demo ...")
-    def run_cmd = "java -cp ${jar_name} org.apache.doris.spark.testcase.TestStreamLoadForArrowType $context.config.feHttpAddress $context.config.feHttpUser regression_test_connector_p0_spark_connector"
+
+    def javaPath = ["bash", "-c", "which java"].execute().text.trim()
+    logger.info("System java path: ${javaPath}")
+
+    def javaVersion = System.getProperty("java.version")
+    logger.info("System java version: ${javaVersion}")
+    def addOpens = ""
+    if (javaVersion.startsWith("17")) {
+        addOpens = "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED  --add-opens=java.base/java.nio=ALL-UNNAMED"
+    }
+
+    def run_cmd = "${javaPath} ${addOpens} -cp ${jar_name} org.apache.doris.spark.testcase.TestStreamLoadForArrowType $context.config.feHttpAddress $context.config.feHttpUser regression_test_connector_p0_spark_connector"
     logger.info("run_cmd : $run_cmd")
     def proc = run_cmd.execute()
     def sout = new StringBuilder()

regression-test/suites/connector_p0/spark_connector/spark_connector_read_type.groovy

@@ -97,7 +97,18 @@ suite("spark_connector_read_type", "connector") {
     logger.info("getS3Url ==== ${getS3Url()}")
     def download_spark_jar = "/usr/bin/curl ${getS3Url()}/regression/spark-doris-read-jar-with-dependencies.jar --output spark-doris-read.jar".execute().getText()
     logger.info("finish download spark doris demo ...")
-    def run_cmd = "java -jar spark-doris-read.jar $context.config.feHttpAddress $context.config.feHttpUser regression_test_connector_p0_spark_connector.$tableReadName regression_test_connector_p0_spark_connector.$tableWriterName"
+
+    def javaPath = ["bash", "-c", "which java"].execute().text.trim()
+    logger.info("System java path: ${javaPath}")
+
+    def javaVersion = System.getProperty("java.version")
+    logger.info("System java version: ${javaVersion}")
+    def addOpens = ""
+    if (javaVersion.startsWith("17")) {
+        addOpens = "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED  --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED"
+    }
+
+    def run_cmd = "${javaPath} ${addOpens} -jar spark-doris-read.jar $context.config.feHttpAddress $context.config.feHttpUser regression_test_connector_p0_spark_connector.$tableReadName regression_test_connector_p0_spark_connector.$tableWriterName"
     logger.info("run_cmd : $run_cmd")
     def proc = run_cmd.execute()
     def sout = new StringBuilder()
@@ -109,7 +120,4 @@ suite("spark_connector_read_type", "connector") {
     }
 
     qt_select """ select * from $tableWriterName order by id"""
-
-
-
 }

regression-test/suites/flink_connector_p0/flink_connector.groovy

@@ -42,7 +42,13 @@ suite("flink_connector") {
         return
     }
 
-    def run_cmd = "java -cp flink-doris-demo.jar com.doris.DorisFlinkDfSinkDemo $context.config.feHttpAddress regression_test_flink_connector_p0.$tableName $context.config.feHttpUser"
+    def javaPath = ["bash", "-c", "which java"].execute().text.trim()
+    logger.info("System java path: ${javaPath}")
+
+    def javaVersion = System.getProperty("java.version")
+    logger.info("System java version: ${javaVersion}")
+
+    def run_cmd = "${javaPath} -cp flink-doris-demo.jar com.doris.DorisFlinkDfSinkDemo $context.config.feHttpAddress regression_test_flink_connector_p0.$tableName $context.config.feHttpUser"
     logger.info("run_cmd : $run_cmd")
     def run_flink_jar = run_cmd.execute().getText()
     logger.info("result: $run_flink_jar")

regression-test/suites/flink_connector_p0/flink_connector_syncdb.groovy

@@ -74,7 +74,17 @@ PROPERTIES (
         throw new Exception("File flink-doris-syncdb.jar download failed.")
     }
 
-    def run_cmd = "java -cp flink-doris-syncdb.jar org.apache.doris.DatabaseFullSync $context.config.feHttpAddress regression_test_flink_connector_p0 $context.config.feHttpUser"
+    def javaPath = ["bash", "-c", "which java"].execute().text.trim()
+    logger.info("System java path: ${javaPath}")
+    def javaVersion = System.getProperty("java.version")
+    logger.info("System java version: ${javaVersion}")
+
+    def addOpens = ""
+    if (javaVersion.startsWith("17")) {
+        addOpens = "--add-opens=java.base/java.nio=ALL-UNNAMED --add-opens java.base/java.lang=ALL-UNNAMED"
+    }
+
+    def run_cmd = "${javaPath} ${addOpens} -cp flink-doris-syncdb.jar org.apache.doris.DatabaseFullSync $context.config.feHttpAddress regression_test_flink_connector_p0 $context.config.feHttpUser"
     logger.info("run_cmd : $run_cmd")
     def run_flink_jar = run_cmd.execute().getText()
     logger.info("result: $run_flink_jar")
@@ -85,7 +95,7 @@ PROPERTIES (
                 logger.info("retry $tableName1  count: $resultTbl1")
                 def resultTbl2 = sql """ select count(1) from $tableName2"""
                 logger.info("retry $tableName2 count: $resultTbl2")
-                resultTbl1.size() >= 1 && resultTbl2.size >=1
+                resultTbl1.size() >= 1 && resultTbl2.size() >=1
             })
 
     qt_select """ select * from $tableName1 order by id"""

regression-test/suites/flink_connector_p0/flink_connector_type.groovy

@@ -19,11 +19,8 @@
 // /testing/trino-product-tests/src/main/resources/sql-tests/testcases/tpcds
 // and modified by Doris.
 
-
-import org.apache.flink.api.common.RuntimeExecutionMode
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
-import org.apache.flink.table.api.TableResult
-import org.apache.flink.table.api.bridge.java.StreamTableEnvironment
+import static java.util.concurrent.TimeUnit.SECONDS
+import org.awaitility.Awaitility
 
 suite("flink_connector_type") {
 
@@ -112,77 +109,44 @@ VALUES
     def thisDb = sql """select database()""";
     thisDb = thisDb[0][0];
     logger.info("current database is ${thisDb}");
-    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-    env.setParallelism(1);
-    env.setRuntimeMode(RuntimeExecutionMode.BATCH);
-    final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
 
-    tEnv.executeSql(
-            "CREATE TABLE source_doris (" +
-                    "`id` int,\n" +
-                    "`c1` boolean,\n" +
-                    "`c2` tinyint,\n" +
-                    "`c3` smallint,\n" +
-                    "`c4` int,\n" +
-                    "`c5` bigint,\n" +
-                    "`c6` string,\n" +
-                    "`c7` float,\n" +
-                    "`c8` double,\n" +
-                    "`c9` decimal(12,4),\n" +
-                    "`c10` date,\n" +
-                    "`c11` TIMESTAMP,\n" +
-                    "`c12` char(1),\n" +
-                    "`c13` varchar(256),\n" +
-                    "`c14` Array<String>,\n" +
-                    "`c15` Map<String, String>,\n" +
-                    "`c16` ROW<name String, age int>,\n" +
-                    "`c17` STRING,\n" +
-                    "`c18` STRING"
-                    + ") "
-                    + "WITH (\n"
-                    + "  'connector' = 'doris',\n"
-                    + "  'fenodes' = '" + context.config.feHttpAddress + "',\n"
-                    + "  'table.identifier' = '${thisDb}.test_types_source',\n"
-                    + "  'username' = '" + context.config.feHttpUser + "',\n"
-                    + "  'password' = '" + context.config.feHttpPassword +  "'\n"
-                    + ")");
+    logger.info("start delete local flink-doris-case.jar....")
+    def delete_local_flink_jar = "rm -rf flink-doris-case.jar".execute()
+    logger.info("start download regression/flink-doris-case.jar ....")
+    logger.info("getS3Url: ${getS3Url()}")
+    def download_flink_jar = "wget --quiet --continue --tries=5 ${getS3Url()}/regression/flink-doris-case.jar".execute().getText()
 
+    def file = new File('flink-doris-case.jar')
+    if (file.exists()) {
+        def fileSize = file.length()
+        logger.info("finish download flink-doris-case.jar, size " + fileSize)
+    } else {
+        logger.info("flink-doris-case.jar download failed")
+        throw new Exception("File flink-doris-case.jar download failed.")
+    }
 
-    tEnv.executeSql(
-            "CREATE TABLE doris_test_sink (" +
-                    "`id` int,\n" +
-                    "`c1` boolean,\n" +
-                    "`c2` tinyint,\n" +
-                    "`c3` smallint,\n" +
-                    "`c4` int,\n" +
-                    "`c5` bigint,\n" +
-                    "`c6` string,\n" +
-                    "`c7` float,\n" +
-                    "`c8` double,\n" +
-                    "`c9` decimal(12,4),\n" +
-                    "`c10` date,\n" +
-                    "`c11` TIMESTAMP,\n" +
-                    "`c12` char(1),\n" +
-                    "`c13` varchar(256),\n" +
-                    "`c14` Array<String>,\n" +
-                    "`c15` Map<String, String>,\n" +
-                    "`c16` ROW<name String, age int>,\n" +
-                    "`c17` STRING,\n" +
-                    "`c18` STRING"
-                    + ") "
-                    + "WITH (\n"
-                    + "  'connector' = 'doris',\n"
-                    + "  'fenodes' = '" + context.config.feHttpAddress + "',\n"
-                    + "  'table.identifier' = '${thisDb}.test_types_sink',\n"
-                    + "  'username' = '" + context.config.feHttpUser + "',\n"
-                    + "  'password' = '" + context.config.feHttpPassword +  "',\n"
-                    + "  'sink.properties.format' = 'json',\n"
-                    + "  'sink.properties.read_json_by_line' = 'true',\n"
-                    + "  'sink.label-prefix' = 'label" + UUID.randomUUID() + "'"
-                    + ")");
+    def javaPath = ["bash", "-c", "which java"].execute().text.trim()
+    logger.info("System java path: ${javaPath}")
+    def javaVersion = System.getProperty("java.version")
+    logger.info("System java version: ${javaVersion}")
+
+    def addOpens = ""
+    if (javaVersion.startsWith("17")) {
+        addOpens = "--add-opens=java.base/java.nio=ALL-UNNAMED  --add-opens java.base/java.lang=ALL-UNNAMED"
+    }
+
+    def run_cmd = "${javaPath} ${addOpens} -cp flink-doris-case.jar org.apache.doris.FlinkConnectorTypeCase $context.config.feHttpAddress regression_test_flink_connector_p0 $context.config.feHttpUser"
+    logger.info("run_cmd : $run_cmd")
+    def run_flink_jar = run_cmd.execute().getText()
+    logger.info("result: $run_flink_jar")
+    // The publish in the commit phase is asynchronous
+    Awaitility.await().atMost(30, SECONDS).pollInterval(1, SECONDS).await().until(
+            {
+                def resultTbl = sql """ select count(1) from test_types_sink"""
+                logger.info("retry test_types_sink  count: $resultTbl")
+                resultTbl.size() >= 1
+            })
 
-    TableResult tableResult = tEnv.executeSql("INSERT INTO doris_test_sink select * from source_doris");
-    tableResult.await();
     logger.info("flink job execute finished.");
     qt_select """ select * from test_types_sink order by id"""
 }

run-regression-test.sh

@@ -275,4 +275,4 @@ if [[ "${ONLY_COMPILE}" -eq 0 ]]; then
         -jar ${RUN_JAR:+${RUN_JAR}} \
         -cf "${CONFIG_FILE}" \
         ${REGRESSION_OPTIONS_PREFIX:+${REGRESSION_OPTIONS_PREFIX}} "$@"
-fi
+fi