You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kyuubi.apache.org by ch...@apache.org on 2026/08/13 17:24:25 UTC
(kyuubi) branch master updated: [KYUUBI #6995] Support Flink 2.0, 2.1, 2.2 and 2.3
This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kyuubi.git
The following commit(s) were added to refs/heads/master by this push:
new 3df4e50ce5 [KYUUBI #6995] Support Flink 2.0, 2.1, 2.2 and 2.3
3df4e50ce5 is described below
commit 3df4e50ce54127f38d8efd1f1b19018df345e577
Author: Cheng Pan <ch...@apache.org>
AuthorDate: Fri Aug 14 01:24:14 2026 +0800
[KYUUBI #6995] Support Flink 2.0, 2.1, 2.2 and 2.3
### Why are the changes needed?
Closes #6995.
Extend the Flink SQL engine to support Flink 2.0, 2.1, 2.2 and 2.3, while keeping the existing
1.17 to 1.20 support. The engine keeps shipping a single jar that runs against all supported
Flink versions, so the version differences are bridged by reflection rather than by per-version
source trees.
Flink 2.x changes that the engine has to adapt to:
- FLINK-14068 (2.0.0) removed `org.apache.flink.api.common.time.Time` in favor of `Duration`,
which is a parameter of `EmbeddedJobClient`'s constructor.
- FLINK-33212 (2.0.0) added a `Configuration` parameter to `EmbeddedExecutor`'s constructor.
- FLINK-38974 (2.3.0) split `EmbeddedExecutorFactory`'s and `EmbeddedExecutor`'s job ids into the
application, suspended and terminal ones, and made the dispatcher reject jobs that carry no
registered application id. Flink resolves the factory constructor at compile time, so Kyuubi's
copy of `EmbeddedExecutorFactory` declares both constructors and stamps the application id
captured from the bootstrap job on every submitted `StreamGraph`, which enables application
mode on Flink 2.3.
- FLINK-35625 and FLINK-36310 (2.0.0) merged `flink run-application` into `flink run` and removed
the former, so `FlinkProcessBuilder` picks the action by the version detected from the
`flink-dist` jar under `$FLINK_HOME/lib`.
- FLINK-36760 (2.0.0) removed `CliOptionsParser#checkFilePath`, which is now inlined into
`FlinkEngineUtils`. Note that `DefaultContext#load` needs no adaption, its dependency list
changed from `List<URL>` to `List<URI>` but both erase to the same descriptor, and Flink still
reads the elements as `URL` in `SessionContext#create`.
- `Configuration#getString(ConfigOption)` was removed, `Configuration#get(ConfigOption)` is used
instead, which exists in all supported versions.
Flink 2.x requires Java 11 or higher and is compiled with Java 11 target, so the `flink-2.x`
profiles raise `enforcer.maxJdkVersion` to 11, and the Flink 2.x CI jobs run on Java 17.
### How was this patch tested?
New UTs cover the Flink version detection and the application mode command selection in
`FlinkProcessBuilderSuite`.
`FlinkOperationSuite` asserts the result of `ENCODE` by version, since FLINK-38062 (2.2.0)
corrected its return type from `BINARY` to `VARBINARY`, before that the result was silently
truncated to the first byte.
`externals/kyuubi-flink-sql-engine` tests were run against each Flink binary distribution, with
`FLINK_HOME` pinned to the matching distribution:
| Flink | Result |
| --- | --- |
| 1.20.5 | 111 passed |
| 2.0.2 | 111 passed |
| 2.1.3 | 111 passed |
| 2.2.1 | 111 passed |
| 2.3.0 | 111 passed |
`FlinkOperationOnYarnSuite` covers application mode on Flink 2.3, and the cross-version job
compiles the engine against Flink 1.20 and runs it on the Flink 2.3 distribution.
The GA `flink-it` matrix now always compiles against the default Flink version and varies only
the runtime distribution, covering 1.17, 1.18, 1.19 on Java 8 and 2.0, 2.1, 2.2, 2.3 on Java 17.
### Was this patch authored or co-authored using generative AI tooling?
Assisted-by: Claude Opus 5, Codex
Closes #7628 from pan3793/kyuubi-flink-2.
Closes #6995
9b53c59fc [Cheng Pan] [KYUUBI #6995][FLINK] Support Flink 2.3 application mode
8d711e294 [Cheng Pan] Cancel Flink IT application mode suite per test on Flink 2.3
7fbe4ec1c [Cheng Pan] Skip Flink IT application mode suite on Flink 2.3
e7166614f [Cheng Pan] Fix Flink IT compile against Flink 2.x
a0056ce26 [Cheng Pan] [KYUUBI #6995][FLINK] Support Flink 2.0, 2.1, 2.2 and 2.3
Authored-by: Cheng Pan <ch...@apache.org>
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.github/workflows/master.yml | 16 ++
docs/quick_start/quick_start.rst | 2 +-
.../executors/EmbeddedExecutorFactory.java | 222 +++++++++++++++++++--
.../kyuubi/engine/flink/FlinkEngineUtils.scala | 13 +-
.../kyuubi/engine/flink/FlinkSQLEngine.scala | 2 +-
.../engine/flink/WithFlinkSQLEngineLocal.scala | 4 +-
.../engine/flink/WithFlinkSQLEngineOnYarn.scala | 5 +-
.../flink/operation/FlinkOperationSuite.scala | 23 ++-
.../WithKyuubiServerAndFlinkMiniCluster.scala | 4 +-
.../kyuubi/engine/flink/FlinkProcessBuilder.scala | 30 ++-
.../engine/flink/FlinkProcessBuilderSuite.scala | 32 ++-
pom.xml | 32 +++
12 files changed, 351 insertions(+), 34 deletions(-)
diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml
index ccee35f59c..f3cd8cdf51 100644
--- a/.github/workflows/master.yml
+++ b/.github/workflows/master.yml
@@ -339,6 +339,22 @@ jobs:
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'
+ comment: 'verify-on-flink-2.0-binary'
+ - java: 17
+ flink: '1.20'
+ flink-archive: '-Dflink.archive.mirror=https://www.apache.org/dyn/closer.lua/flink/flink-2.1.3 -Dflink.archive.name=flink-2.1.3-bin-scala_2.12.tgz'
+ comment: 'verify-on-flink-2.1-binary'
+ - java: 17
+ flink: '1.20'
+ flink-archive: '-Dflink.archive.mirror=https://www.apache.org/dyn/closer.lua/flink/flink-2.2.1 -Dflink.archive.name=flink-2.2.1-bin-scala_2.12.tgz'
+ comment: 'verify-on-flink-2.2-binary'
+ - java: 17
+ flink: '1.20'
+ flink-archive: '-Dflink.archive.mirror=https://www.apache.org/dyn/closer.lua/flink/flink-2.3.0 -Dflink.archive.name=flink-2.3.0-bin-scala_2.12.tgz'
+ comment: 'verify-on-flink-2.3-binary'
steps:
- uses: actions/checkout@v7
- name: Free up disk space
diff --git a/docs/quick_start/quick_start.rst b/docs/quick_start/quick_start.rst
index 3e71beaeca..bd97cfec56 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.3 to 3.5, 4.0 to 4.2 A Spark distribution
- **Flink** Engine 1.17 to 1.20 A Flink distribution
+ **Flink** Engine 1.17 to 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/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutorFactory.java b/externals/kyuubi-flink-sql-engine/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutorFactory.java
index 558db74a37..cdf8747703 100644
--- a/externals/kyuubi-flink-sql-engine/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutorFactory.java
+++ b/externals/kyuubi-flink-sql-engine/src/main/java/org/apache/flink/client/deployment/application/executors/EmbeddedExecutorFactory.java
@@ -20,30 +20,120 @@ package org.apache.flink.client.deployment.application.executors;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
+import java.time.Duration;
import java.util.Collection;
+import java.util.Optional;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.time.Time;
import org.apache.flink.client.cli.ClientOptions;
import org.apache.flink.client.deployment.application.EmbeddedJobClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
+import org.apache.flink.core.execution.JobClient;
import org.apache.flink.core.execution.PipelineExecutor;
import org.apache.flink.core.execution.PipelineExecutorFactory;
import org.apache.flink.runtime.dispatcher.DispatcherGateway;
+import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.util.concurrent.ScheduledExecutor;
+import org.apache.kyuubi.util.reflect.DynClasses;
+import org.apache.kyuubi.util.reflect.DynConstructors;
+import org.apache.kyuubi.util.reflect.DynMethods;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/** Copied from Apache Flink to exposed the DispatcherGateway for Kyuubi statements. */
+/**
+ * Copied from Apache Flink to expose the DispatcherGateway for Kyuubi statements and stamp the
+ * application id on submitted StreamGraphs, which is required since FLINK-38974 (2.3.0).
+ */
@Internal
public class EmbeddedExecutorFactory implements PipelineExecutorFactory {
+ /** FLINK-14068 (2.0.0) removed {@code Time} in favor of {@link Duration}. */
+ private static final Class<?> LEGACY_TIME_CLASS =
+ DynClasses.builder().impl("org.apache.flink.api.common.time.Time").orNull().build();
+
+ /** FLINK-38974 (2.3.0) introduced {@code ApplicationID}, absent in older Flink versions. */
+ private static final Class<?> APPLICATION_ID_CLASS =
+ DynClasses.builder().impl("org.apache.flink.api.common.ApplicationID").orNull().build();
+
+ private static final boolean IS_FLINK_1 = LEGACY_TIME_CLASS != null;
+
+ /**
+ * EmbeddedExecutor's constructor changed twice, FLINK-33212 (2.0.0) added the {@link
+ * Configuration} parameter, FLINK-38974 (2.3.0) split the job ids into the application, suspended
+ * and terminal ones. Bind all variants reflectively so that a single engine jar runs on all
+ * supported Flink versions.
+ */
+ private static final DynConstructors.Ctor<EmbeddedExecutor> EMBEDDED_EXECUTOR_CTOR =
+ DynConstructors.builder()
+ .impl(
+ EmbeddedExecutor.class,
+ Collection.class,
+ Collection.class,
+ Collection.class,
+ DispatcherGateway.class,
+ Configuration.class,
+ EmbeddedJobClientCreator.class)
+ .impl(
+ EmbeddedExecutor.class,
+ Collection.class,
+ DispatcherGateway.class,
+ Configuration.class,
+ EmbeddedJobClientCreator.class)
+ .impl(
+ EmbeddedExecutor.class,
+ Collection.class,
+ DispatcherGateway.class,
+ EmbeddedJobClientCreator.class)
+ .build();
+
+ /** FLINK-14068 (2.0.0) replaced the {@code Time} timeout parameter with {@link Duration}. */
+ private static final DynConstructors.Ctor<EmbeddedJobClient> EMBEDDED_JOB_CLIENT_CTOR =
+ DynConstructors.builder()
+ .impl(
+ EmbeddedJobClient.class,
+ JobID.class,
+ DispatcherGateway.class,
+ ScheduledExecutor.class,
+ Duration.class,
+ ClassLoader.class)
+ .impl(
+ EmbeddedJobClient.class,
+ JobID.class,
+ DispatcherGateway.class,
+ ScheduledExecutor.class,
+ LEGACY_TIME_CLASS,
+ ClassLoader.class)
+ .build();
+
+ private static final DynMethods.StaticMethod LEGACY_TIME_OF_MILLIS =
+ IS_FLINK_1
+ ? DynMethods.builder("milliseconds").impl(LEGACY_TIME_CLASS, long.class).buildStatic()
+ : null;
+
private static Collection<JobID> bootstrapJobIds;
private static Collection<JobID> submittedJobIds;
+ /**
+ * Both are null before FLINK-38974 (2.3.0), which is also how the Flink version is told apart.
+ */
+ private static Collection<JobID> suspendedJobIds;
+
+ private static Collection<JobID> terminalJobIds;
+
+ /**
+ * FLINK-38974 (2.3.0): the dispatcher rejects jobs that carry no application registered in it.
+ * Kyuubi statements run through the SQL gateway, which builds plain StreamExecutionEnvironments,
+ * so capture the id from the first StreamGraph that carries it and stamp it on the rest.
+ */
+ private static volatile Object applicationId;
+
+ private static volatile DynMethods.UnboundMethod streamGraphGetApplicationId;
+
+ private static volatile DynMethods.UnboundMethod streamGraphSetApplicationId;
+
private static DispatcherGateway dispatcherGateway;
private static ScheduledExecutor retryExecutor;
@@ -65,7 +155,7 @@ public class EmbeddedExecutorFactory implements PipelineExecutorFactory {
}
/**
- * Creates an {@link EmbeddedExecutorFactory}.
+ * Creates an {@link EmbeddedExecutorFactory}, invoked by Flink before FLINK-38974 (2.3.0).
*
* @param submittedJobIds a list that is going to be filled with the job ids of the new jobs that
* will be submitted. This is essentially used to return the submitted job ids to the caller.
@@ -76,6 +166,28 @@ public class EmbeddedExecutorFactory implements PipelineExecutorFactory {
final Collection<JobID> submittedJobIds,
final DispatcherGateway dispatcherGateway,
final ScheduledExecutor retryExecutor) {
+ this(submittedJobIds, null, null, dispatcherGateway, retryExecutor);
+ }
+
+ /**
+ * Creates an {@link EmbeddedExecutorFactory}, invoked by Flink since FLINK-38974 (2.3.0). Flink
+ * resolves this constructor at compile time, so it has to exist even though Kyuubi never calls it
+ * directly.
+ *
+ * @param applicationJobIds a list that is going to be filled with the job ids of the new jobs
+ * that will be submitted. This is essentially used to return the submitted job ids to the
+ * caller.
+ * @param suspendedJobIds ids of jobs suspended by a previous application execution.
+ * @param terminalJobIds ids of jobs already terminated by a previous application execution.
+ * @param dispatcherGateway the dispatcher of the cluster which is going to be used to submit
+ * jobs.
+ */
+ public EmbeddedExecutorFactory(
+ final Collection<JobID> applicationJobIds,
+ final Collection<JobID> suspendedJobIds,
+ final Collection<JobID> terminalJobIds,
+ final DispatcherGateway dispatcherGateway,
+ final ScheduledExecutor retryExecutor) {
// there should be only one instance of EmbeddedExecutorFactory
LOGGER.debug(
"{} initiated in thread {} with classloader {}.",
@@ -90,8 +202,10 @@ public class EmbeddedExecutorFactory implements PipelineExecutorFactory {
// issues
LOGGER.debug("Bootstrapping EmbeddedExecutorFactory.");
EmbeddedExecutorFactory.submittedJobIds =
- new ConcurrentLinkedQueue<>(checkNotNull(submittedJobIds));
- EmbeddedExecutorFactory.bootstrapJobIds = submittedJobIds;
+ new ConcurrentLinkedQueue<>(checkNotNull(applicationJobIds));
+ EmbeddedExecutorFactory.bootstrapJobIds = applicationJobIds;
+ EmbeddedExecutorFactory.suspendedJobIds = suspendedJobIds;
+ EmbeddedExecutorFactory.terminalJobIds = terminalJobIds;
EmbeddedExecutorFactory.dispatcherGateway = checkNotNull(dispatcherGateway);
EmbeddedExecutorFactory.retryExecutor = checkNotNull(retryExecutor);
bootstrapLock.notifyAll();
@@ -141,14 +255,94 @@ public class EmbeddedExecutorFactory implements PipelineExecutorFactory {
LOGGER.info("Bootstrapping Flink SQL engine with the initial SQL.");
executorJobIDs = bootstrapJobIds;
}
- return new EmbeddedExecutor(
- executorJobIDs,
- dispatcherGateway,
- (jobId, userCodeClassloader) -> {
- final Time timeout =
- Time.milliseconds(configuration.get(ClientOptions.CLIENT_TIMEOUT).toMillis());
- return new EmbeddedJobClient(
- jobId, dispatcherGateway, retryExecutor, timeout, userCodeClassloader);
- });
+ final EmbeddedJobClientCreator jobClientCreator =
+ (jobId, userCodeClassloader) ->
+ newEmbeddedJobClient(
+ jobId, configuration.get(ClientOptions.CLIENT_TIMEOUT), userCodeClassloader);
+ return stampApplicationId(newEmbeddedExecutor(executorJobIDs, configuration, jobClientCreator));
+ }
+
+ private static PipelineExecutor newEmbeddedExecutor(
+ final Collection<JobID> jobIds,
+ final Configuration configuration,
+ final EmbeddedJobClientCreator jobClientCreator) {
+ if (suspendedJobIds != null) {
+ return EMBEDDED_EXECUTOR_CTOR.newInstance(
+ jobIds,
+ suspendedJobIds,
+ terminalJobIds,
+ dispatcherGateway,
+ configuration,
+ jobClientCreator);
+ }
+ return IS_FLINK_1
+ ? EMBEDDED_EXECUTOR_CTOR.newInstance(jobIds, dispatcherGateway, jobClientCreator)
+ : EMBEDDED_EXECUTOR_CTOR.newInstance(
+ jobIds, dispatcherGateway, configuration, jobClientCreator);
+ }
+
+ /**
+ * FLINK-38974 (2.3.0) requires submitted jobs to carry the id of the application registered in
+ * the dispatcher. The bootstrap SQL goes through StreamContextEnvironment which stamps the id,
+ * while Kyuubi statements go through plain environments, so cache the id and stamp it on every
+ * StreamGraph before delegating to the {@link EmbeddedExecutor}.
+ */
+ private static PipelineExecutor stampApplicationId(final PipelineExecutor executor) {
+ if (suspendedJobIds == null) {
+ return executor;
+ }
+ return (pipeline, configuration, userCodeClassloader) -> {
+ if (pipeline instanceof StreamGraph) {
+ final StreamGraph streamGraph = (StreamGraph) pipeline;
+ final Object appId = captureApplicationId(streamGraph);
+ if (appId != null) {
+ streamGraphSetApplicationId(streamGraph, appId);
+ }
+ }
+ return executor.execute(pipeline, configuration, userCodeClassloader);
+ };
+ }
+
+ private static Object captureApplicationId(final StreamGraph streamGraph) {
+ Object appId = applicationId;
+ if (appId == null) {
+ synchronized (bootstrapLock) {
+ appId = applicationId;
+ if (appId == null) {
+ initStreamGraphApplicationIdMethods();
+ applicationId = appId = streamGraphGetApplicationId(streamGraph);
+ }
+ }
+ }
+ return appId;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Object streamGraphGetApplicationId(final StreamGraph streamGraph) {
+ return ((Optional<Object>) streamGraphGetApplicationId.invoke(streamGraph)).orElse(null);
+ }
+
+ private static void streamGraphSetApplicationId(
+ final StreamGraph streamGraph, final Object applicationId) {
+ streamGraphSetApplicationId.invoke(streamGraph, applicationId);
+ }
+
+ private static void initStreamGraphApplicationIdMethods() {
+ if (streamGraphGetApplicationId == null) {
+ streamGraphGetApplicationId =
+ DynMethods.builder("getApplicationId").impl(StreamGraph.class).build();
+ streamGraphSetApplicationId =
+ DynMethods.builder("setApplicationId")
+ .impl(StreamGraph.class, APPLICATION_ID_CLASS)
+ .build();
+ }
+ }
+
+ private static JobClient newEmbeddedJobClient(
+ final JobID jobId, final Duration timeout, final ClassLoader userCodeClassloader) {
+ final Object rpcTimeout =
+ IS_FLINK_1 ? LEGACY_TIME_OF_MILLIS.invoke(timeout.toMillis()) : timeout;
+ return EMBEDDED_JOB_CLIENT_CTOR.newInstance(
+ jobId, dispatcherGateway, retryExecutor, rpcTimeout, userCodeClassloader);
}
}
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 8b2b1237fd..991ebb12c4 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
@@ -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").map(SemanticVersion.apply)
+ Set("1.17", "1.18", "1.19", "1.20", "2.0", "2.1", "2.2", "2.3").map(SemanticVersion.apply)
val FLINK_RUNTIME_VERSION: SemanticVersion = SemanticVersion(EnvironmentInformation.getVersion)
@@ -172,6 +172,17 @@ object FlinkEngineUtils extends Logging {
} else null
}
+ /**
+ * Copied from [[org.apache.flink.table.client.cli.CliOptionsParser]], which dropped the method
+ * in Flink 2.0.
+ */
+ private def checkFilePath(filePath: String): Unit = {
+ val scheme = new Path(filePath).toUri.getScheme
+ if (scheme != null && scheme != "file") {
+ throw new SqlClientException("SQL Client only supports to load files in local.")
+ }
+ }
+
def renewDelegationToken(delegationToken: String): Unit = {
val newCreds = KyuubiHadoopUtils.decodeCredentials(delegationToken)
val newTokens = KyuubiHadoopUtils.getTokenMap(newCreds)
diff --git a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkSQLEngine.scala b/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkSQLEngine.scala
index bc772ced8f..8e37bd5ddc 100644
--- a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkSQLEngine.scala
+++ b/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkSQLEngine.scala
@@ -94,7 +94,7 @@ object FlinkSQLEngine extends Logging {
.toMap
flinkConf.addAll(Configuration.fromMap(flinkConfFromArgs.asJava))
- val executionTarget = flinkConf.getString(DeploymentOptions.TARGET)
+ val executionTarget = flinkConf.get(DeploymentOptions.TARGET)
setDeploymentConf(executionTarget, flinkConf)
kyuubiConf.setIfMissing(KyuubiConf.FRONTEND_THRIFT_BINARY_BIND_PORT, 0)
diff --git a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngineLocal.scala b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngineLocal.scala
index a73ca27d3c..33dac8cab5 100644
--- a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngineLocal.scala
+++ b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngineLocal.scala
@@ -187,8 +187,8 @@ trait WithFlinkSQLEngineLocal extends KyuubiFunSuite with WithFlinkTestResources
.build
miniCluster = new MiniCluster(cfg)
miniCluster.start()
- flinkConfig.setString(RestOptions.ADDRESS, miniCluster.getRestAddress.get().getHost)
- flinkConfig.setInteger(RestOptions.PORT, miniCluster.getRestAddress.get().getPort)
+ flinkConfig.set(RestOptions.ADDRESS, miniCluster.getRestAddress.get().getHost)
+ flinkConfig.set(RestOptions.PORT, Int.box(miniCluster.getRestAddress.get().getPort))
}
protected def getJdbcUrl: String = s"jdbc:hive2://$connectionUrl/;"
diff --git a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngineOnYarn.scala b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngineOnYarn.scala
index 3690046e6c..2ebbaeea63 100644
--- a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngineOnYarn.scala
+++ b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngineOnYarn.scala
@@ -33,6 +33,7 @@ import org.apache.hadoop.yarn.server.MiniYARNCluster
import org.apache.kyuubi.{KYUUBI_VERSION, KyuubiFunSuite, SCALA_COMPILE_VERSION, Utils}
import org.apache.kyuubi.config.KyuubiConf
import org.apache.kyuubi.config.KyuubiConf.{ENGINE_FLINK_APPLICATION_JARS, KYUUBI_HOME_ENV_VAR_NAME}
+import org.apache.kyuubi.engine.flink.FlinkEngineUtils.FLINK_RUNTIME_VERSION
import org.apache.kyuubi.ha.HighAvailabilityConf.HA_ADDRESSES
import org.apache.kyuubi.util.JavaUtils
import org.apache.kyuubi.util.command.CommandLineUtils._
@@ -167,7 +168,9 @@ trait WithFlinkSQLEngineOnYarn extends KyuubiFunSuite with WithFlinkTestResource
val command = new ArrayBuffer[String]()
command += s"${envs("FLINK_HOME")}${File.separator}bin/flink"
- command += "run-application"
+ // Flink 2.0 merged `run-application` into `run` (FLINK-35625) and removed the former
+ // (FLINK-36310)
+ command += (if (FLINK_RUNTIME_VERSION < "2.0") "run-application" else "run")
command += "-t"
command += "yarn-application"
command += s"-Dyarn.ship-files=${flinkExtraJars.mkString(";")}"
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 4bba523aff..f8e93badad 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
@@ -866,8 +866,10 @@ abstract class FlinkOperationSuite extends HiveJDBCTestHelper with WithFlinkTest
withJdbcStatement() { statement =>
val resultSet = statement.executeQuery("select encode('kyuubi', 'UTF-8')")
assert(resultSet.next())
- // TODO: validate table results after FLINK-28882 is resolved
- assert(resultSet.getString(1) == "k")
+ // FLINK-38062 (2.2.0) corrected the return type of ENCODE from BINARY to VARBINARY, before
+ // that the result was silently truncated to the first byte
+ val expected = if (FLINK_RUNTIME_VERSION < "2.2") "k" else "kyuubi"
+ assert(resultSet.getString(1) === expected)
val metaData = resultSet.getMetaData
assert(metaData.getColumnType(1) === java.sql.Types.BINARY)
}
@@ -1269,16 +1271,19 @@ abstract class FlinkOperationSuite extends HiveJDBCTestHelper with WithFlinkTest
}
test("test result fetch timeout") {
- val exception = intercept[KyuubiSQLException](
- withSessionConf()(Map(ENGINE_FLINK_FETCH_TIMEOUT.key -> "PT60S"))() {
- withJdbcStatement("tbl_a") { stmt =>
- stmt.executeQuery("create table tbl_a (a int) " +
- "with ('connector' = 'datagen', 'rows-per-second'='0')")
+ withSessionConf()(Map(ENGINE_FLINK_FETCH_TIMEOUT.key -> "PT60S"))() {
+ withJdbcStatement("tbl_a") { stmt =>
+ stmt.executeQuery("create table tbl_a (a int) " +
+ "with ('connector' = 'datagen', 'rows-per-second'='0')")
+ // only the fetch is expected to time out, keep the session setup out of the intercept so
+ // that it does not swallow the assumptions of the inheriting suites
+ val exception = intercept[KyuubiSQLException] {
val resultSet = stmt.executeQuery("select * from tbl_a")
while (resultSet.next()) {}
}
- })
- assert(exception.getMessage === "Futures timed out after [60000 milliseconds]")
+ assert(exception.getMessage === "Futures timed out after [60000 milliseconds]")
+ }
+ }
}
test("execute statement - help") {
diff --git a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/WithKyuubiServerAndFlinkMiniCluster.scala b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/WithKyuubiServerAndFlinkMiniCluster.scala
index 5ac991fcef..35e84d5246 100644
--- a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/WithKyuubiServerAndFlinkMiniCluster.scala
+++ b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/WithKyuubiServerAndFlinkMiniCluster.scala
@@ -47,7 +47,7 @@ trait WithKyuubiServerAndFlinkMiniCluster extends WithKyuubiServer {
.build
miniCluster = new MiniCluster(cfg)
miniCluster.start()
- flinkConfig.setString(RestOptions.ADDRESS, miniCluster.getRestAddress.get().getHost)
- flinkConfig.setInteger(RestOptions.PORT, miniCluster.getRestAddress.get().getPort)
+ flinkConfig.set(RestOptions.ADDRESS, miniCluster.getRestAddress.get().getHost)
+ flinkConfig.set(RestOptions.PORT, Int.box(miniCluster.getRestAddress.get().getPort))
}
}
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 20817842d0..ae1ed14a1c 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
@@ -35,7 +35,7 @@ import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_ENGINE_CREDENTIALS_KE
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
+import org.apache.kyuubi.util.{KyuubiHadoopUtils, SemanticVersion}
import org.apache.kyuubi.util.command.CommandLineUtils._
/**
@@ -63,6 +63,16 @@ class FlinkProcessBuilder(
// flink.execution.target are required in Kyuubi conf currently
val executionTarget: Option[String] = conf.getOption("flink.execution.target")
+ private[kyuubi] lazy val flinkVersion: SemanticVersion = {
+ val libDir = Paths.get(flinkHome, "lib").toFile
+ val fileNames = Option(libDir.list()).getOrElse {
+ throw new KyuubiException(
+ s"Failed to list jars under $flinkHome, please check if FLINK_HOME is configured " +
+ "correctly and the lib directory exists")
+ }
+ extractFlinkVersion(fileNames)
+ }
+
private lazy val proxyUserEnable: Boolean = {
var flinkDoAsEnabled = conf.get(ENGINE_FLINK_DOAS_ENABLED)
if (flinkDoAsEnabled && !UserGroupInformation.isSecurityEnabled) {
@@ -107,7 +117,7 @@ class FlinkProcessBuilder(
case Some("yarn-application") =>
val buffer = new mutable.ListBuffer[String]()
buffer += flinkExecutable
- buffer += "run-application"
+ buffer += runApplicationCommand(flinkVersion)
val flinkExtraJars = new mutable.ListBuffer[String]
// locate flink sql jars
@@ -323,4 +333,20 @@ object FlinkProcessBuilder {
final val FLINK_PROXY_USER_KEY = "HADOOP_PROXY_USER"
final val FLINK_SECURITY_KEYTAB_KEY = "security.kerberos.login.keytab"
final val FLINK_SECURITY_PRINCIPAL_KEY = "security.kerberos.login.principal"
+
+ final private[kyuubi] val FLINK_DIST_VERSION_REGEX =
+ """^flink-dist-(\d+\.\d+)[.-].*\.jar$""".r
+
+ /**
+ * Flink 2.0 merged `run-application` into `run` (FLINK-35625) and removed the former
+ * (FLINK-36310).
+ */
+ final private[kyuubi] def runApplicationCommand(flinkVersion: SemanticVersion): String =
+ if (flinkVersion < "2.0") "run-application" else "run"
+
+ final private[kyuubi] def extractFlinkVersion(fileNames: Iterable[String]): SemanticVersion = {
+ Option(fileNames).getOrElse(Iterable.empty)
+ .collectFirst { case FLINK_DIST_VERSION_REGEX(version) => SemanticVersion(version) }
+ .getOrElse(throw new KyuubiException("Failed to extract Flink version from flink-dist jar"))
+ }
}
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 ed02a7db41..c11465b2a8 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
@@ -24,13 +24,14 @@ import scala.collection.JavaConverters._
import scala.collection.immutable.ListMap
import scala.util.matching.Regex
-import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.{KyuubiException, KyuubiFunSuite}
import org.apache.kyuubi.config.KyuubiConf
import org.apache.kyuubi.config.KyuubiConf.{ENGINE_FLINK_APPLICATION_JARS, ENGINE_FLINK_EXTRA_CLASSPATH, ENGINE_FLINK_JAVA_OPTIONS, ENGINE_FLINK_MEMORY}
import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_ENGINE_APP_MGR_INFO_KEY, KYUUBI_ENGINE_CREDENTIALS_KEY}
import org.apache.kyuubi.engine.{ApplicationManagerInfo, EngineType}
import org.apache.kyuubi.engine.ApplicationManagerInfo.serialize
import org.apache.kyuubi.engine.flink.FlinkProcessBuilder._
+import org.apache.kyuubi.util.SemanticVersion
class FlinkProcessBuilderSuite extends KyuubiFunSuite {
private def sessionModeConf = KyuubiConf()
@@ -56,6 +57,9 @@ class FlinkProcessBuilderSuite extends KyuubiFunSuite {
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"))
+ 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"))
private val tempUsrLib =
Files.createDirectories(Paths.get(tempFlinkHome.toPath.toString, "usrlib")).toFile
private val tempUdfJar =
@@ -201,4 +205,30 @@ class FlinkProcessBuilderSuite extends KyuubiFunSuite {
val matcher = regex.pattern.matcher(actualCommands)
assert(matcher.matches())
}
+
+ test("extract flink version from flink-dist jar") {
+ Seq(
+ "flink-dist-1.20.5.jar" -> "1.20",
+ "flink-dist-1.20.0-rc1.jar" -> "1.20",
+ "flink-dist-2.0.2.jar" -> "2.0",
+ "flink-dist-2.3.0-SNAPSHOT.jar" -> "2.3").foreach { case (jar, expected) =>
+ assertResult(SemanticVersion(expected))(extractFlinkVersion(Seq(jar)))
+ }
+
+ Seq(
+ "flink-dist_2.12-1.14.6.jar",
+ "flink-dist-1.20.5.zip",
+ "flink-table-runtime-2.0.2.jar").foreach { jar =>
+ assertThrows[KyuubiException](extractFlinkVersion(Seq(jar)))
+ }
+ }
+
+ test("application mode uses run-application only before flink 2.0") {
+ Seq("1.17", "1.19", "1.20").foreach { version =>
+ assertResult("run-application")(runApplicationCommand(SemanticVersion(version)))
+ }
+ Seq("2.0", "2.1", "2.2", "2.3").foreach { version =>
+ assertResult("run")(runApplicationCommand(SemanticVersion(version)))
+ }
+ }
}
diff --git a/pom.xml b/pom.xml
index efb34a1341..5361ddea6f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2191,6 +2191,38 @@
</properties>
</profile>
+ <profile>
+ <id>flink-2.0</id>
+ <properties>
+ <flink.version>2.0.2</flink.version>
+ <enforcer.maxJdkVersion>11</enforcer.maxJdkVersion>
+ </properties>
+ </profile>
+
+ <profile>
+ <id>flink-2.1</id>
+ <properties>
+ <flink.version>2.1.3</flink.version>
+ <enforcer.maxJdkVersion>11</enforcer.maxJdkVersion>
+ </properties>
+ </profile>
+
+ <profile>
+ <id>flink-2.2</id>
+ <properties>
+ <flink.version>2.2.1</flink.version>
+ <enforcer.maxJdkVersion>11</enforcer.maxJdkVersion>
+ </properties>
+ </profile>
+
+ <profile>
+ <id>flink-2.3</id>
+ <properties>
+ <flink.version>2.3.0</flink.version>
+ <enforcer.maxJdkVersion>11</enforcer.maxJdkVersion>
+ </properties>
+ </profile>
+
<profile>
<id>zookeeper-3.6</id>
<properties>