Drop support for Flink 1.17, 1.18 and 1.19

.github/workflows/master.yml

@@ -297,23 +297,10 @@ jobs:
         java:
           - 8
         flink:
-          - '1.19'
           - '1.20'
         flink-archive: [ "" ]
         comment: [ "normal" ]
         include:
-          - java: 8
-            flink: '1.20'
-            flink-archive: '-Dflink.archive.mirror=https://www.apache.org/dyn/closer.lua/flink/flink-1.17.2 -Dflink.archive.name=flink-1.17.2-bin-scala_2.12.tgz'
-            comment: 'verify-on-flink-1.17-binary'
-          - java: 8
-            flink: '1.20'
-            flink-archive: '-Dflink.archive.mirror=https://www.apache.org/dyn/closer.lua/flink/flink-1.18.1 -Dflink.archive.name=flink-1.18.1-bin-scala_2.12.tgz'
-            comment: 'verify-on-flink-1.18-binary'
-          - java: 8
-            flink: '1.20'
-            flink-archive: '-Dflink.archive.mirror=https://www.apache.org/dyn/closer.lua/flink/flink-1.19.3 -Dflink.archive.name=flink-1.19.3-bin-scala_2.12.tgz'
-            comment: 'verify-on-flink-1.19-binary'
           - java: 17
             flink: '1.20'
             flink-archive: '-Dflink.archive.mirror=https://www.apache.org/dyn/closer.lua/flink/flink-2.0.2 -Dflink.archive.name=flink-2.0.2-bin-scala_2.12.tgz'

AGENTS.md

@@ -70,8 +70,8 @@ Use `-am` (also-make) when building or testing a single module — without it, M
 | Profile                                                 | Notes                                        |
 |---------------------------------------------------------|----------------------------------------------|
 | `-Pspark-3.5` (default), `-Pspark-{4.0,4.1,4.2,master}` | Spark version                                |
-| `-Pflink-1.20` (default), `-Pflink-{1.17,1.18,1.19}`    | Flink version                                |
-| `-Pscala-2.13`                                          | Scala 2.13 (default is 2.12)                 |
+| `-Pflink-1.20` (default), `-Pflink-{2.0,2.1,2.2,2.3}`   | Flink version                                |
+| `-Pscala-2.12` (default), `-Pscala-2.13`                | Scala version                                |
 | `-P{spark,flink,hive}-provided`                         | skip bundled engine downloads                |
 | `-Pmirror-cdn`                                          | use Apache mirror CDN for engine archives    |
 | `-Pfast`                                                | skip tests/style/docs/enforcer/RAT/downloads |

docs/connector/flink/hudi.rst

@@ -44,7 +44,7 @@ The **classpath** of kyuubi flink sql engine with Hudi supported consists of
 
 1. kyuubi-flink-sql-engine-\ |release|\ _2.12.jar, the engine jar deployed with a Kyuubi distribution
 2. a copy of flink distribution
-3. hudi-flink<flink.version>-bundle-<hudi.version>.jar (example: hudi-flink1.18-bundle-1.0.1.jar), which can be found in the `Maven Central`_
+3. hudi-flink<flink.version>-bundle-<hudi.version>.jar (example: hudi-flink1.20-bundle-1.0.1.jar), which can be found in the `Maven Central`_
 
 In order to make the Hudi packages visible for the runtime classpath of engines, we can use one of these methods:
 

docs/connector/flink/iceberg.rst

@@ -45,7 +45,7 @@ The **classpath** of kyuubi flink sql engine with Iceberg supported consists of
 
 1. kyuubi-flink-sql-engine-\ |release|\ _2.12.jar, the engine jar deployed with a Kyuubi distribution
 2. a copy of flink distribution
-3. iceberg-flink-runtime-<flink.version>-<iceberg.version>.jar (example: iceberg-flink-runtime-1.18-1.7.0.jar), which can be found in the `Maven Central`_
+3. iceberg-flink-runtime-<flink.version>-<iceberg.version>.jar (example: iceberg-flink-runtime-1.20-1.7.0.jar), which can be found in the `Maven Central`_
 
 In order to make the Iceberg packages visible for the runtime classpath of engines, we can use one of these methods:
 

docs/connector/flink/paimon.rst

@@ -42,7 +42,7 @@ The **classpath** of kyuubi flink sql engine with Apache Paimon supported consis
 
 1. kyuubi-flink-sql-engine-\ |release|\ _2.12.jar, the engine jar deployed with a Kyuubi distribution
 2. a copy of flink distribution
-3. paimon-flink-<version>.jar (example: paimon-flink-1.18-0.8.1.jar), which can be found in the `Apache Paimon Supported Engines Flink`_
+3. paimon-flink-<version>.jar (example: paimon-flink-1.20-0.8.1.jar), which can be found in the `Apache Paimon Supported Engines Flink`_
 4. flink-shaded-hadoop-2-uber-<version>.jar, which code can be found in the `Pre-bundled Hadoop Jar`_
 
 In order to make the Apache Paimon packages visible for the runtime classpath of engines, you need to:

docs/deployment/engine_on_yarn.md

@@ -129,7 +129,7 @@ To deploy Kyuubi's Flink SQL engines on YARN, you'd better have cognition upon t
 
 ### Flink Deployment Modes
 
-Currently, Flink supports two deployment modes on YARN: [YARN Application Mode](https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/resource-providers/yarn/#application-mode) and [YARN Session Mode](https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/resource-providers/yarn/#application-mode).
+Currently, Flink supports two deployment modes on YARN: [YARN Application Mode](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/resource-providers/yarn/#application-mode) and [YARN Session Mode](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/resource-providers/yarn/#application-mode).
 
 - YARN Application Mode: In this mode, Kyuubi starts a dedicated Flink application cluster and runs the SQL engine on it.
 - YARN Session Mode: In this mode, Kyuubi starts the Flink SQL engine locally and connects to a running Flink YARN session cluster.

docs/deployment/migration-guide.md

@@ -20,6 +20,8 @@
 ## Upgrading from Kyuubi 1.12 to 1.13
 
 * Since Kyuubi 1.13, the support of Spark engine for Spark 3.3 and 3.4 is removed.
+* Since Kyuubi 1.13, the support of Flink engine for Flink 1.17, 1.18 and 1.19 is removed.
+* Since Kyuubi 1.13, the support of Flink engine for Flink 2.0 is deprecated, and will be removed in the future.
 
 ## Upgrading from Kyuubi 1.11 to 1.12
 

docs/extensions/engines/flink/functions.md

@@ -18,7 +18,7 @@
 # Auxiliary SQL Functions
 
 Kyuubi provides several auxiliary SQL functions as supplement to
-Flink's [Built-in Functions](https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/functions/systemfunctions/)
+Flink's [Built-in Functions](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/functions/systemfunctions/)
 
 |        Name         |                         Description                         | Return Type | Since |
 |---------------------|-------------------------------------------------------------|-------------|-------|

docs/quick_start/quick_start.rst

@@ -44,7 +44,7 @@ pre-installed and the ``JAVA_HOME`` is correctly set to each component.
                    Engine lib                            - Kyuubi Engine
                    Beeline                               - Kyuubi Beeline
   **Spark**        Engine       3.5, 4.0 to 4.2          A Spark distribution
-  **Flink**        Engine       1.17 to 1.20, 2.0 to 2.3 A Flink distribution
+  **Flink**        Engine       1.20, 2.0 to 2.3         A Flink distribution
   **Trino**        Engine       N/A                      A Trino cluster allows to access via trino-client v411
   **Doris**        Engine       N/A                      A Doris cluster
   **Hive**         Engine       - 2.1-cdh6, 2.3, 3.1     - A Hive distribution

externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkEngineUtils.scala

@@ -38,7 +38,7 @@ import org.apache.flink.table.gateway.service.session.Session
 import org.apache.flink.util.JarUtils
 import org.apache.hadoop.security.{Credentials, UserGroupInformation}
 
-import org.apache.kyuubi.{KyuubiException, Logging}
+import org.apache.kyuubi.Logging
 import org.apache.kyuubi.util.{KyuubiHadoopUtils, SemanticVersion}
 import org.apache.kyuubi.util.reflect.ReflectUtils._
 
@@ -47,16 +47,17 @@ object FlinkEngineUtils extends Logging {
   val EMBEDDED_MODE_CLIENT_OPTIONS: Options = getEmbeddedModeClientOptions(new Options)
 
   private def SUPPORTED_FLINK_VERSIONS =
-    Set("1.17", "1.18", "1.19", "1.20", "2.0", "2.1", "2.2", "2.3").map(SemanticVersion.apply)
+    Set("1.20", "2.0", "2.1", "2.2", "2.3").map(SemanticVersion.apply)
 
   val FLINK_RUNTIME_VERSION: SemanticVersion = SemanticVersion(EnvironmentInformation.getVersion)
 
   def checkFlinkVersion(): Unit = {
     val flinkVersion = EnvironmentInformation.getVersion
     if (SUPPORTED_FLINK_VERSIONS.contains(FLINK_RUNTIME_VERSION)) {
       info(s"The current Flink version is $flinkVersion")
-      if (FlinkEngineUtils.FLINK_RUNTIME_VERSION <= "1.19") {
-        warn("The support for Flink 1.17, 1.18 and 1.19 is deprecated, and will be removed " +
+      if (FlinkEngineUtils.FLINK_RUNTIME_VERSION.majorVersion == 2 &&
+        FlinkEngineUtils.FLINK_RUNTIME_VERSION.minorVersion == 0) {
+        warn("The support for Flink 2.0 is deprecated, and will be removed " +
           "in a future version.")
       }
     } else {
@@ -111,25 +112,14 @@ object FlinkEngineUtils extends Logging {
     val libDirs: JList[URL] = Option(checkUrls(line, CliOptionsParser.OPTION_LIBRARY))
       .getOrElse(JCollections.emptyList())
     val dependencies: JList[URL] = discoverDependencies(jars, libDirs)
-    if (FLINK_RUNTIME_VERSION >= "1.19") {
-      invokeAs[DefaultContext](
-        classOf[DefaultContext],
-        "load",
-        (classOf[Configuration], flinkConf),
-        (classOf[JList[URL]], dependencies),
-        (classOf[Boolean], JBoolean.TRUE))
-    } else if (FLINK_RUNTIME_VERSION >= "1.17") {
-      invokeAs[DefaultContext](
-        classOf[DefaultContext],
-        "load",
-        (classOf[Configuration], flinkConf),
-        (classOf[JList[URL]], dependencies),
-        (classOf[Boolean], JBoolean.TRUE),
-        (classOf[Boolean], JBoolean.FALSE))
-    } else {
-      throw new KyuubiException(
-        s"Flink version ${EnvironmentInformation.getVersion} are not supported currently.")
-    }
+    // FLINK-36760 (2.0.0) changed the second parameter from List<URL> to List<URI>, use
+    // reflection to bridge Flink 1.20 and 2.x
+    invokeAs[DefaultContext](
+      classOf[DefaultContext],
+      "load",
+      (classOf[Configuration], flinkConf),
+      (classOf[JList[URL]], dependencies),
+      (classOf[Boolean], JBoolean.TRUE))
   }
 
   def getSessionContext(session: Session): SessionContext = getField(session, "sessionContext")

externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/result/IncrementalResultFetchIterator.scala

@@ -35,7 +35,6 @@ import org.apache.flink.table.types.DataType
 import org.apache.flink.types.Row
 
 import org.apache.kyuubi.Logging
-import org.apache.kyuubi.engine.flink.shim.FlinkResultSet
 import org.apache.kyuubi.operation.FetchIterator
 import org.apache.kyuubi.util.reflect.DynFields
 
@@ -88,8 +87,7 @@ class IncrementalResultFetchIterator(
       debug(s"Fetching from result store with timeout $resultFetchTimeout ms")
       while (!fetched && !Thread.interrupted()) {
         val rs = resultFetcher.fetchResults(token, effectiveMaxRows - bufferedRows.length)
-        val flinkRs = new FlinkResultSet(rs)
-        flinkRs.getResultType match {
+        rs.getResultType match {
           case ResultType.EOS =>
             debug("EOS received, no more data to fetch.")
             fetched = true
@@ -98,7 +96,7 @@ class IncrementalResultFetchIterator(
             // if flink jobs are not ready, continue to retry
             debug("Result not ready, retrying...")
           case ResultType.PAYLOAD =>
-            val fetchedData = flinkRs.getData
+            val fetchedData = rs.getData
             // if no data fetched, continue to retry
             if (!fetchedData.isEmpty) {
               debug(s"Fetched ${fetchedData.length} rows from result store.")
@@ -109,10 +107,10 @@ class IncrementalResultFetchIterator(
               debug("No data fetched, retrying...")
             }
           case _ =>
-            throw new RuntimeException(s"Unexpected result type: ${flinkRs.getResultType}")
+            throw new RuntimeException(s"Unexpected result type: ${rs.getResultType}")
         }
         if (hasNext) {
-          val nextToken = flinkRs.getNextToken
+          val nextToken = rs.getNextToken
           if (nextToken == null) {
             hasNext = false
           } else {

externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationSuite.scala

@@ -636,58 +636,56 @@ abstract class FlinkOperationSuite extends HiveJDBCTestHelper with WithFlinkTest
   }
 
   test("execute statement - show/stop jobs") {
-    if (FLINK_RUNTIME_VERSION >= "1.17") {
-      // use a bigger value to ensure all tasks of the streaming query run until
-      // we explicitly stop the job.
-      withSessionConf()(Map(ENGINE_FLINK_MAX_ROWS.key -> "10000"))(Map.empty) {
-        withMultipleConnectionJdbcStatement()({ statement =>
-          statement.executeQuery(
-            "create table tbl_a (a int) with (" +
-              "'connector' = 'datagen', " +
-              "'rows-per-second'='10')")
-          statement.executeQuery("create table tbl_b (a int) with ('connector' = 'blackhole')")
-          val insertResult1 = statement.executeQuery("insert into tbl_b select * from tbl_a")
-          assert(insertResult1.next())
-          val jobId1 = insertResult1.getString(1)
-
-          Thread.sleep(5000)
-
-          val showResult = statement.executeQuery("show jobs")
-          val metadata = showResult.getMetaData
-          assert(metadata.getColumnName(1) === "job id")
-          assert(metadata.getColumnType(1) === java.sql.Types.VARCHAR)
-          assert(metadata.getColumnName(2) === "job name")
-          assert(metadata.getColumnType(2) === java.sql.Types.VARCHAR)
-          assert(metadata.getColumnName(3) === "status")
-          assert(metadata.getColumnType(3) === java.sql.Types.VARCHAR)
-          assert(metadata.getColumnName(4) === "start time")
-          assert(metadata.getColumnType(4) === java.sql.Types.OTHER)
-
-          var isFound = false
-          while (showResult.next()) {
-            if (showResult.getString(1) === jobId1) {
-              isFound = true
-              assert(showResult.getString(2) === "test-job")
-              assert(showResult.getString(3) === "RUNNING")
-              assert(showResult.getObject(4).isInstanceOf[TimestampTZ])
-            }
+    // use a bigger value to ensure all tasks of the streaming query run until
+    // we explicitly stop the job.
+    withSessionConf()(Map(ENGINE_FLINK_MAX_ROWS.key -> "10000"))(Map.empty) {
+      withMultipleConnectionJdbcStatement()({ statement =>
+        statement.executeQuery(
+          "create table tbl_a (a int) with (" +
+            "'connector' = 'datagen', " +
+            "'rows-per-second'='10')")
+        statement.executeQuery("create table tbl_b (a int) with ('connector' = 'blackhole')")
+        val insertResult1 = statement.executeQuery("insert into tbl_b select * from tbl_a")
+        assert(insertResult1.next())
+        val jobId1 = insertResult1.getString(1)
+
+        Thread.sleep(5000)
+
+        val showResult = statement.executeQuery("show jobs")
+        val metadata = showResult.getMetaData
+        assert(metadata.getColumnName(1) === "job id")
+        assert(metadata.getColumnType(1) === java.sql.Types.VARCHAR)
+        assert(metadata.getColumnName(2) === "job name")
+        assert(metadata.getColumnType(2) === java.sql.Types.VARCHAR)
+        assert(metadata.getColumnName(3) === "status")
+        assert(metadata.getColumnType(3) === java.sql.Types.VARCHAR)
+        assert(metadata.getColumnName(4) === "start time")
+        assert(metadata.getColumnType(4) === java.sql.Types.OTHER)
+
+        var isFound = false
+        while (showResult.next()) {
+          if (showResult.getString(1) === jobId1) {
+            isFound = true
+            assert(showResult.getString(2) === "test-job")
+            assert(showResult.getString(3) === "RUNNING")
+            assert(showResult.getObject(4).isInstanceOf[TimestampTZ])
           }
-          assert(isFound)
+        }
+        assert(isFound)
 
-          val stopResult1 = statement.executeQuery(s"stop job '$jobId1'")
-          assert(stopResult1.next())
-          assert(stopResult1.getString(1) === "OK")
+        val stopResult1 = statement.executeQuery(s"stop job '$jobId1'")
+        assert(stopResult1.next())
+        assert(stopResult1.getString(1) === "OK")
 
-          val insertResult2 = statement.executeQuery("insert into tbl_b select * from tbl_a")
-          assert(insertResult2.next())
-          val jobId2 = insertResult2.getString(1)
+        val insertResult2 = statement.executeQuery("insert into tbl_b select * from tbl_a")
+        assert(insertResult2.next())
+        val jobId2 = insertResult2.getString(1)
 
-          val stopResult2 = statement.executeQuery(s"stop job '$jobId2' with savepoint")
-          assert(stopResult2.getMetaData.getColumnName(1).equals("savepoint path"))
-          assert(stopResult2.next())
-          assert(Paths.get(stopResult2.getString(1)).getFileName.toString.startsWith("savepoint-"))
-        })
-      }
+        val stopResult2 = statement.executeQuery(s"stop job '$jobId2' with savepoint")
+        assert(stopResult2.getMetaData.getColumnName(1).equals("savepoint path"))
+        assert(stopResult2.next())
+        assert(Paths.get(stopResult2.getString(1)).getFileName.toString.startsWith("savepoint-"))
+      })
     }
   }
 
@@ -1060,11 +1058,9 @@ abstract class FlinkOperationSuite extends HiveJDBCTestHelper with WithFlinkTest
       val jobId = resultSet.getString(1)
       assert(jobId.length == 32)
 
-      if (FLINK_RUNTIME_VERSION >= "1.17") {
-        val stopResult = statement.executeQuery(s"stop job '$jobId'")
-        assert(stopResult.next())
-        assert(stopResult.getString(1) === "OK")
-      }
+      val stopResult = statement.executeQuery(s"stop job '$jobId'")
+      assert(stopResult.next())
+      assert(stopResult.getString(1) === "OK")
     })
   }
 
@@ -1152,20 +1148,18 @@ abstract class FlinkOperationSuite extends HiveJDBCTestHelper with WithFlinkTest
         assert(rows === 200)
       }
     }
-    if (FLINK_RUNTIME_VERSION >= "1.17") {
-      withSessionConf()(Map(ENGINE_FLINK_MAX_ROWS.key -> "10"))(Map.empty) {
-        withJdbcStatement() { statement =>
-          for (i <- 0 to 10) {
-            statement.execute(s"create table tbl_src$i (a bigint) " +
-              s"with ('connector' = 'blackhole')")
-          }
-          val resultSet = statement.executeQuery("show tables")
-          var rows = 0
-          while (resultSet.next()) {
-            rows += 1
-          }
-          assert(rows === 11)
+    withSessionConf()(Map(ENGINE_FLINK_MAX_ROWS.key -> "10"))(Map.empty) {
+      withJdbcStatement() { statement =>
+        for (i <- 0 to 10) {
+          statement.execute(s"create table tbl_src$i (a bigint) " +
+            s"with ('connector' = 'blackhole')")
+        }
+        val resultSet = statement.executeQuery("show tables")
+        var rows = 0
+        while (resultSet.next()) {
+          rows += 1
         }
+        assert(rows === 11)
       }
     }
   }