From a33970f57db709cb22726503475af5bbe4f02e22 Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Fri, 14 Aug 2026 14:45:59 +0800 Subject: [PATCH 1/2] Drop support for Flink 1.17, 1.18 and 1.19 ### Why are the changes needed? Flink 1.17, 1.18 and 1.19 are EOL and were deprecated earlier (KYUUBI #7199, KYUUBI #7285), with their building support already removed. Kyuubi 1.13 drops the remaining support, making Flink 1.20 the minimum. Flink 2.0 is EOL, so it is marked deprecated. ### What changes are proposed? - Remove the flink-1.17/flink-1.18/flink-1.19 Maven profiles and their deprecation warning - Drop Flink 1.17/1.18/1.19 from the CI matrix - Clean up stale adaptation code that only existed for Flink < 1.20, including the FlinkResultSet reflection shim, the 4-arg DefaultContext.load branch, and >= "1.17" test guards - Deprecate Flink 2.0 at engine startup - Update docs to record the removal and deprecation ### How was this patch tested? Cross-version tested: engine compiled with Flink 1.20, run against a Flink 2.3.0 distribution; FlinkOperationSuite and FlinkOperationSuiteOnYarn passed 12/12. CI coverage: the flink-it matrix keeps Flink 1.20 (JDK 8) plus Flink 2.0/2.1/2.2/2.3 binary verification (JDK 17). Spotless and Scalastyle pass. ### Was this patch authored or co-authored using generative AI tooling? Assisted-by: DeepSeek V4 Pro --- .github/workflows/master.yml | 13 -- AGENTS.md | 4 +- docs/connector/flink/hudi.rst | 2 +- docs/connector/flink/iceberg.rst | 2 +- docs/connector/flink/paimon.rst | 2 +- docs/deployment/engine_on_yarn.md | 2 +- docs/deployment/migration-guide.md | 2 + docs/extensions/engines/flink/functions.md | 2 +- docs/quick_start/quick_start.rst | 2 +- .../engine/flink/FlinkEngineUtils.scala | 36 ++--- .../IncrementalResultFetchIterator.scala | 10 +- .../engine/flink/shim/FlinkResultSet.scala | 35 ----- .../flink/operation/FlinkOperationSuite.scala | 126 +++++++++--------- .../flink/FlinkProcessBuilderSuite.scala | 6 +- pom.xml | 24 ---- 15 files changed, 90 insertions(+), 178 deletions(-) delete mode 100644 externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/shim/FlinkResultSet.scala diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml index 999f84400e2..f071a034fc5 100644 --- a/.github/workflows/master.yml +++ b/.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' diff --git a/AGENTS.md b/AGENTS.md index 3e6901a6259..1c94c974cc8 100644 --- a/AGENTS.md +++ b/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 | diff --git a/docs/connector/flink/hudi.rst b/docs/connector/flink/hudi.rst index 4d766eb2efe..e27749ace21 100644 --- a/docs/connector/flink/hudi.rst +++ b/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-bundle-.jar (example: hudi-flink1.18-bundle-1.0.1.jar), which can be found in the `Maven Central`_ +3. hudi-flink-bundle-.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: diff --git a/docs/connector/flink/iceberg.rst b/docs/connector/flink/iceberg.rst index 7b956365e01..ab85160e83a 100644 --- a/docs/connector/flink/iceberg.rst +++ b/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--.jar (example: iceberg-flink-runtime-1.18-1.7.0.jar), which can be found in the `Maven Central`_ +3. iceberg-flink-runtime--.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: diff --git a/docs/connector/flink/paimon.rst b/docs/connector/flink/paimon.rst index e960df0fce5..ff7ffc52df3 100644 --- a/docs/connector/flink/paimon.rst +++ b/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-.jar (example: paimon-flink-1.18-0.8.1.jar), which can be found in the `Apache Paimon Supported Engines Flink`_ +3. paimon-flink-.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-.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: diff --git a/docs/deployment/engine_on_yarn.md b/docs/deployment/engine_on_yarn.md index 01401d199cd..f5b14e9c56b 100644 --- a/docs/deployment/engine_on_yarn.md +++ b/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. diff --git a/docs/deployment/migration-guide.md b/docs/deployment/migration-guide.md index bdcadd87609..38fcda6e53d 100644 --- a/docs/deployment/migration-guide.md +++ b/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 diff --git a/docs/extensions/engines/flink/functions.md b/docs/extensions/engines/flink/functions.md index 1d047d07889..0111823c878 100644 --- a/docs/extensions/engines/flink/functions.md +++ b/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 | |---------------------|-------------------------------------------------------------|-------------|-------| diff --git a/docs/quick_start/quick_start.rst b/docs/quick_start/quick_start.rst index eacc6f18673..98e1ba8648b 100644 --- a/docs/quick_start/quick_start.rst +++ b/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 diff --git a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkEngineUtils.scala b/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkEngineUtils.scala index 991ebb12c48..2f0b0710d1d 100644 --- a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkEngineUtils.scala +++ b/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,7 +47,7 @@ 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) @@ -55,8 +55,9 @@ object FlinkEngineUtils extends Logging { 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 to List, 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") diff --git a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/result/IncrementalResultFetchIterator.scala b/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/result/IncrementalResultFetchIterator.scala index b43f6fa224f..be79cd8efa5 100644 --- a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/result/IncrementalResultFetchIterator.scala +++ b/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 { diff --git a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/shim/FlinkResultSet.scala b/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/shim/FlinkResultSet.scala deleted file mode 100644 index 7fb05c8446b..00000000000 --- a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/shim/FlinkResultSet.scala +++ /dev/null @@ -1,35 +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. - */ - -package org.apache.kyuubi.engine.flink.shim - -import java.lang.{Long => JLong} -import java.util - -import org.apache.flink.table.data.RowData -import org.apache.flink.table.gateway.api.results.ResultSet.ResultType - -import org.apache.kyuubi.util.reflect.ReflectUtils._ - -class FlinkResultSet(resultSet: AnyRef) { - - def getData: util.List[RowData] = invokeAs(resultSet, "getData") - - def getNextToken: JLong = invokeAs(resultSet, "getNextToken") - - def getResultType: ResultType = invokeAs(resultSet, "getResultType") -} diff --git a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationSuite.scala b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationSuite.scala index f8e93badad1..60c44c991ee 100644 --- a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationSuite.scala +++ b/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) } } } diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilderSuite.scala index c11465b2a85..d8dc0072630 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilderSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilderSuite.scala @@ -55,11 +55,11 @@ class FlinkProcessBuilderSuite extends KyuubiFunSuite { private val tempFlinkHome = Files.createTempDirectory("flink-home").toFile private val tempOpt = Files.createDirectories(Paths.get(tempFlinkHome.toPath.toString, "opt")).toFile - Files.createFile(Paths.get(tempOpt.toPath.toString, "flink-sql-client-1.17.2.jar")) - Files.createFile(Paths.get(tempOpt.toPath.toString, "flink-sql-gateway-1.17.2.jar")) + Files.createFile(Paths.get(tempOpt.toPath.toString, "flink-sql-client-1.20.5.jar")) + Files.createFile(Paths.get(tempOpt.toPath.toString, "flink-sql-gateway-1.20.5.jar")) private val tempLib = Files.createDirectories(Paths.get(tempFlinkHome.toPath.toString, "lib")).toFile - Files.createFile(Paths.get(tempLib.toPath.toString, "flink-dist-1.17.2.jar")) + Files.createFile(Paths.get(tempLib.toPath.toString, "flink-dist-1.20.5.jar")) private val tempUsrLib = Files.createDirectories(Paths.get(tempFlinkHome.toPath.toString, "usrlib")).toFile private val tempUdfJar = diff --git a/pom.xml b/pom.xml index a7d02e9f759..8e9a00db08f 100644 --- a/pom.xml +++ b/pom.xml @@ -2125,30 +2125,6 @@ - - - flink-1.17 - - 1.17.2 - - - - - flink-1.18 - - 1.18.1 - - - - - flink-1.19 - - 1.19.3 - - - flink-1.20 From 76c03d987ef2fbfce81fbff59cd475379a676008 Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Fri, 14 Aug 2026 17:22:36 +0800 Subject: [PATCH 2/2] Remove obsolete kyuubi.engine.flink.doAs.generateTokenFile FLINK-35525 (1.20.0) allows the kyuubi delegation token provider to pass tokens to the YARN AM, and Flink 1.20 is now the minimum supported version, so the temporary HADOOP_TOKEN_FILE workaround is no longer needed. --- .../org/apache/kyuubi/config/KyuubiConf.scala | 11 ----- .../engine/flink/FlinkProcessBuilder.scala | 49 +++---------------- 2 files changed, 8 insertions(+), 52 deletions(-) diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala index 4239f8b6187..4e97e71659d 100644 --- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala +++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala @@ -3462,17 +3462,6 @@ object KyuubiConf { .booleanConf .createWithDefault(false) - val ENGINE_FLINK_DOAS_GENERATE_TOKEN_FILE: ConfigEntry[Boolean] = - buildConf("kyuubi.engine.flink.doAs.generateTokenFile") - .internal - .doc(s"When ${ENGINE_FLINK_DOAS_ENABLED.key}=true and neither FLINK-35525 (Flink 1.20.0)" + - " nor YARN-10333 (Hadoop 3.4.0) is available, enable this configuration to generate a" + - " temporary HADOOP_TOKEN_FILE that will be picked up by the Flink engine bootstrap" + - " process.") - .version("1.10.0") - .booleanConf - .createWithDefault(false) - val SERVER_LIMIT_CONNECTIONS_PER_USER: OptionalConfigEntry[Int] = buildConf("kyuubi.server.limit.connections.per.user") .doc("Maximum kyuubi server connections per user." + diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilder.scala index ae1ed14a1c2..29bc9bfdba6 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilder.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilder.scala @@ -24,18 +24,16 @@ import scala.collection.mutable import com.google.common.annotations.VisibleForTesting import org.apache.commons.lang3.StringUtils -import org.apache.hadoop.conf.Configuration -import org.apache.hadoop.fs.Path import org.apache.hadoop.security.UserGroupInformation import org.apache.kyuubi._ import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf._ -import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_ENGINE_CREDENTIALS_KEY, KYUUBI_SESSION_USER_KEY} +import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_SESSION_USER_KEY import org.apache.kyuubi.engine.{ApplicationManagerInfo, EngineType, KyuubiApplicationManager, ProcBuilder} import org.apache.kyuubi.engine.flink.FlinkProcessBuilder._ import org.apache.kyuubi.operation.log.OperationLog -import org.apache.kyuubi.util.{KyuubiHadoopUtils, SemanticVersion} +import org.apache.kyuubi.util.SemanticVersion import org.apache.kyuubi.util.command.CommandLineUtils._ /** @@ -92,7 +90,7 @@ class FlinkProcessBuilder( val flinkExtraEnvs = if (proxyUserEnable) { Map( "FLINK_CONF_DIR" -> flinkConfDir, - FLINK_PROXY_USER_KEY -> proxyUser) ++ generateTokenFile() + FLINK_PROXY_USER_KEY -> proxyUser) } else { Map("FLINK_CONF_DIR" -> flinkConfDir) } @@ -147,14 +145,14 @@ class FlinkProcessBuilder( val externalProxyUserConf: Map[String, String] = if (proxyUserEnable) { // FLINK-31109 (1.17.0): Flink only supports hadoop proxy user when delegation tokens - // fetch is managed outside, but disabling `security.delegation.tokens.enabled` will cause - // delegation token updates on JobManager not to be passed to TaskManagers. + // fetch is managed outside, but disabling `security.delegation.tokens.enabled` will + // cause delegation token updates on JobManager not to be passed to TaskManagers. // Based on the solution in // https://github.com/apache/flink/pull/22009#issuecomment-2122226755, we removed // `HadoopModuleFactory` from `security.module.factory.classes` and disabled delegation // token providers (hadoopfs/hbase/HiveServer2) that do not support proxyUser. - // FLINK-35525: We need to add `yarn.security.appmaster.delegation.token.services=kyuubi` - // configuration to pass hdfs token obtained by kyuubi provider to the yarn client. + // FLINK-35525 (1.20.0): add `yarn.security.appmaster.delegation.token.services=kyuubi` + // to pass the hdfs token obtained by kyuubi provider to the yarn client. Map( "security.module.factory.classes" -> ("org.apache.flink.runtime.security.modules.JaasModuleFactory;" + @@ -267,7 +265,7 @@ class FlinkProcessBuilder( if (!Files.exists(devHadoopJars)) { throw new KyuubiException( s"The path $devHadoopJars does not exist. Please set " + - s"${FLINK_HADOOP_CLASSPATH_KEY} or ${ENGINE_FLINK_EXTRA_CLASSPATH.key} " + + s"$FLINK_HADOOP_CLASSPATH_KEY or ${ENGINE_FLINK_EXTRA_CLASSPATH.key} " + s"to configure the location of Hadoop client jars. Alternatively," + s"you can place the required hadoop-client or flink-shaded-hadoop jars " + s"directly into the Flink lib directory: $flinkHome/lib.") @@ -287,37 +285,6 @@ class FlinkProcessBuilder( } } - @volatile private var tokenTempDir: java.nio.file.Path = _ - private def generateTokenFile(): Option[(String, String)] = { - if (conf.get(ENGINE_FLINK_DOAS_GENERATE_TOKEN_FILE)) { - // We disabled `hadoopfs` token service, which may cause yarn client to miss hdfs token. - // So we generate a hadoop token file to pass kyuubi engine tokens to submit process. - // TODO: Removed this after FLINK-35525 (1.20.0), delegation tokens will be passed - // by `kyuubi` provider - conf.getOption(KYUUBI_ENGINE_CREDENTIALS_KEY).map { encodedCredentials => - val credentials = KyuubiHadoopUtils.decodeCredentials(encodedCredentials) - tokenTempDir = Utils.createTempDir() - val file = s"${tokenTempDir.toString}/kyuubi_credentials_${System.currentTimeMillis()}" - credentials.writeTokenStorageFile(new Path(s"file://$file"), new Configuration()) - info(s"Generated hadoop token file: $file") - "HADOOP_TOKEN_FILE_LOCATION" -> file - } - } else { - None - } - } - - override def close(destroyProcess: Boolean): Unit = { - super.close(destroyProcess) - if (tokenTempDir != null) { - try { - Utils.deleteDirectoryRecursively(tokenTempDir.toFile) - } catch { - case e: Throwable => error(s"Error deleting token temp dir: $tokenTempDir", e) - } - } - } - override def shortName: String = "flink" }