From a0056ce26ad80cbf632accc8fd59ee5205ab108b Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Wed, 12 Aug 2026 23:32:48 +0800 Subject: [PATCH 1/5] [KYUUBI #6995][FLINK] Support Flink 2.0, 2.1, 2.2 and 2.3 Assisted-by: Claude Opus 5 --- .github/workflows/master.yml | 16 ++ docs/quick_start/quick_start.rst | 2 +- .../executors/EmbeddedExecutorFactory.java | 143 ++++++++++++++++-- .../engine/flink/FlinkEngineUtils.scala | 13 +- .../kyuubi/engine/flink/FlinkSQLEngine.scala | 2 +- .../flink/WithFlinkSQLEngineLocal.scala | 4 +- .../flink/WithFlinkSQLEngineOnYarn.scala | 5 +- .../operation/FlinkOperationOnYarnSuite.scala | 11 +- .../flink/operation/FlinkOperationSuite.scala | 23 +-- .../engine/flink/FlinkProcessBuilder.scala | 30 +++- .../flink/FlinkProcessBuilderSuite.scala | 32 +++- pom.xml | 32 ++++ 12 files changed, 281 insertions(+), 32 deletions(-) diff --git a/.github/workflows/master.yml b/.github/workflows/master.yml index 951c2c20016..7cbe5b9b295 100644 --- a/.github/workflows/master.yml +++ b/.github/workflows/master.yml @@ -334,6 +334,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 3e71beaecaa..bd97cfec562 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 558db74a372..b0f09fd801e 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,19 +20,23 @@ 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.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.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; @@ -40,10 +44,76 @@ @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(); + + 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 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 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 bootstrapJobIds; private static Collection submittedJobIds; + /** + * Both are null before FLINK-38974 (2.3.0), which is also how the Flink version is told apart. + */ + private static Collection suspendedJobIds; + + private static Collection terminalJobIds; + private static DispatcherGateway dispatcherGateway; private static ScheduledExecutor retryExecutor; @@ -65,7 +135,7 @@ public EmbeddedExecutorFactory() { } /** - * 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 +146,28 @@ public EmbeddedExecutorFactory( final Collection 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 applicationJobIds, + final Collection suspendedJobIds, + final Collection terminalJobIds, + final DispatcherGateway dispatcherGateway, + final ScheduledExecutor retryExecutor) { // there should be only one instance of EmbeddedExecutorFactory LOGGER.debug( "{} initiated in thread {} with classloader {}.", @@ -90,8 +182,10 @@ public EmbeddedExecutorFactory( // 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 +235,37 @@ public PipelineExecutor getExecutor(final Configuration configuration) { 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 newEmbeddedExecutor(executorJobIDs, configuration, jobClientCreator); + } + + private static PipelineExecutor newEmbeddedExecutor( + final Collection 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); + } + + 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 8b2b1237fda..991ebb12c48 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 bc772ced8f2..8e37bd5ddcb 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 a73ca27d3c1..33dac8cab5c 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 3690046e6c5..2ebbaeea630 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/FlinkOperationOnYarnSuite.scala b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationOnYarnSuite.scala index 401c3b0bdd0..a513aba14fc 100644 --- a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationOnYarnSuite.scala +++ b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationOnYarnSuite.scala @@ -24,12 +24,21 @@ import org.apache.kyuubi.config.KyuubiConf.{ENGINE_SHARE_LEVEL, ENGINE_TYPE} import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_SESSION_USER_KEY import org.apache.kyuubi.engine.ShareLevel import org.apache.kyuubi.engine.flink.{WithDiscoveryFlinkSQLEngine, WithFlinkSQLEngineOnYarn} +import org.apache.kyuubi.engine.flink.FlinkEngineUtils.FLINK_RUNTIME_VERSION import org.apache.kyuubi.ha.HighAvailabilityConf.{HA_ENGINE_REF_ID, HA_NAMESPACE} class FlinkOperationOnYarnSuite extends FlinkOperationSuite with WithDiscoveryFlinkSQLEngine with WithFlinkSQLEngineOnYarn { - protected def jdbcUrl: String = getFlinkEngineServiceUrl + protected def jdbcUrl: String = { + // FLINK-38974 (2.3.0) rejects jobs that are not associated with an application registered in + // the dispatcher. The engine submits jobs through its own copy of EmbeddedExecutorFactory, + // which does not stamp an application id, so every query fails on Flink 2.3 and later. + assume( + FLINK_RUNTIME_VERSION < "2.3", + "Flink application mode is not supported since Flink 2.3, see FLINK-38974") + getFlinkEngineServiceUrl + } override def withKyuubiConf: Map[String, String] = { Map( 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 4bba523affb..f8e93badad1 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/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 20817842d02..ae1ed14a1c2 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 ed02a7db41c..c11465b2a85 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 efb34a1341c..5361ddea6ff 100644 --- a/pom.xml +++ b/pom.xml @@ -2191,6 +2191,38 @@ + + flink-2.0 + + 2.0.2 + 11 + + + + + flink-2.1 + + 2.1.3 + 11 + + + + + flink-2.2 + + 2.2.1 + 11 + + + + + flink-2.3 + + 2.3.0 + 11 + + + zookeeper-3.6 From e7166614f4d956d3ac26a2660dbdad9ce8ac9e9c Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Thu, 13 Aug 2026 04:14:19 +0800 Subject: [PATCH 2/5] Fix Flink IT compile against Flink 2.x FLINK-14068 (2.0.0) removed Configuration#setString/setInteger, use set --- .../kyuubi/it/flink/WithKyuubiServerAndFlinkMiniCluster.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 5ac991fcefe..35e84d52460 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)) } } From 7fbe4ec1cf600a270220b515d917a0e4632341c4 Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Thu, 13 Aug 2026 07:06:42 +0800 Subject: [PATCH 3/5] Skip Flink IT application mode suite on Flink 2.3 FLINK-38974 (2.3.0) rejects jobs that carry no application id in the dispatcher --- .../it/flink/operation/FlinkOperationSuiteOnYarn.scala | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala index c06b742d7aa..b8a55d8548e 100644 --- a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala +++ b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala @@ -17,12 +17,15 @@ package org.apache.kyuubi.it.flink.operation +import org.apache.flink.runtime.util.EnvironmentInformation + import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf._ import org.apache.kyuubi.it.flink.WithKyuubiServerAndYarnMiniCluster import org.apache.kyuubi.operation.HiveJDBCTestHelper import org.apache.kyuubi.operation.meta.ResultSetSchemaConstant.TABLE_CAT import org.apache.kyuubi.shaded.hive.service.rpc.thrift.{TGetInfoReq, TGetInfoType} +import org.apache.kyuubi.util.SemanticVersion class FlinkOperationSuiteOnYarn extends WithKyuubiServerAndYarnMiniCluster with HiveJDBCTestHelper { @@ -35,6 +38,11 @@ class FlinkOperationSuiteOnYarn extends WithKyuubiServerAndYarnMiniCluster } override def beforeAll(): Unit = { + // FLINK-38974 (2.3.0) rejects jobs that are not associated with an application registered in + // the dispatcher, so application mode does not work on Flink 2.3 and later + assume( + SemanticVersion(EnvironmentInformation.getVersion) < "2.3", + "Flink application mode is not supported since Flink 2.3, see FLINK-38974") conf .set(s"$KYUUBI_ENGINE_ENV_PREFIX.$KYUUBI_HOME_ENV_VAR_NAME", kyuubiHome) .set(ENGINE_TYPE, "FLINK_SQL") From 8d711e294e3a229edd19333d255a27c2545bfae1 Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Thu, 13 Aug 2026 09:47:23 +0800 Subject: [PATCH 4/5] Cancel Flink IT application mode suite per test on Flink 2.3 assume in beforeAll aborts the whole suite and is reported as failure, guard jdbcUrl instead so each test is canceled --- .../it/flink/operation/FlinkOperationSuiteOnYarn.scala | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala index b8a55d8548e..6c8681b8cce 100644 --- a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala +++ b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala @@ -31,6 +31,11 @@ class FlinkOperationSuiteOnYarn extends WithKyuubiServerAndYarnMiniCluster with HiveJDBCTestHelper { override protected def jdbcUrl: String = { + // FLINK-38974 (2.3.0) rejects jobs that are not associated with an application registered in + // the dispatcher, so application mode does not work on Flink 2.3 and later + assume( + SemanticVersion(EnvironmentInformation.getVersion) < "2.3", + "Flink application mode is not supported since Flink 2.3, see FLINK-38974") // delay the access to thrift service because the thrift service // may not be ready although it's registered Thread.sleep(3000L) @@ -38,11 +43,6 @@ class FlinkOperationSuiteOnYarn extends WithKyuubiServerAndYarnMiniCluster } override def beforeAll(): Unit = { - // FLINK-38974 (2.3.0) rejects jobs that are not associated with an application registered in - // the dispatcher, so application mode does not work on Flink 2.3 and later - assume( - SemanticVersion(EnvironmentInformation.getVersion) < "2.3", - "Flink application mode is not supported since Flink 2.3, see FLINK-38974") conf .set(s"$KYUUBI_ENGINE_ENV_PREFIX.$KYUUBI_HOME_ENV_VAR_NAME", kyuubiHome) .set(ENGINE_TYPE, "FLINK_SQL") From 9b53c59fc5bd9721199cf83fbda0eb637ab04546 Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Thu, 13 Aug 2026 21:26:19 +0800 Subject: [PATCH 5/5] [KYUUBI #6995][FLINK] Support Flink 2.3 application mode FLINK-38974 (2.3.0) makes the dispatcher reject jobs that carry no application registered in it. Kyuubi statements run through the SQL gateway, which builds plain StreamExecutionEnvironments, so stamp the application id captured from the bootstrap job on every submitted StreamGraph. Assisted-by: Codex --- .../executors/EmbeddedExecutorFactory.java | 81 ++++++++++++++++++- .../operation/FlinkOperationOnYarnSuite.scala | 11 +-- .../operation/FlinkOperationSuiteOnYarn.scala | 8 -- 3 files changed, 80 insertions(+), 20 deletions(-) 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 b0f09fd801e..cdf87477034 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 @@ -22,6 +22,7 @@ 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; @@ -33,6 +34,7 @@ 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; @@ -40,7 +42,10 @@ 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 { @@ -48,6 +53,10 @@ public class EmbeddedExecutorFactory implements PipelineExecutorFactory { 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; /** @@ -114,6 +123,17 @@ public class EmbeddedExecutorFactory implements PipelineExecutorFactory { private static Collection 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; @@ -239,7 +259,7 @@ public PipelineExecutor getExecutor(final Configuration configuration) { (jobId, userCodeClassloader) -> newEmbeddedJobClient( jobId, configuration.get(ClientOptions.CLIENT_TIMEOUT), userCodeClassloader); - return newEmbeddedExecutor(executorJobIDs, configuration, jobClientCreator); + return stampApplicationId(newEmbeddedExecutor(executorJobIDs, configuration, jobClientCreator)); } private static PipelineExecutor newEmbeddedExecutor( @@ -261,6 +281,63 @@ private static PipelineExecutor newEmbeddedExecutor( 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) 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 = diff --git a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationOnYarnSuite.scala b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationOnYarnSuite.scala index a513aba14fc..401c3b0bdd0 100644 --- a/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationOnYarnSuite.scala +++ b/externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationOnYarnSuite.scala @@ -24,21 +24,12 @@ import org.apache.kyuubi.config.KyuubiConf.{ENGINE_SHARE_LEVEL, ENGINE_TYPE} import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_SESSION_USER_KEY import org.apache.kyuubi.engine.ShareLevel import org.apache.kyuubi.engine.flink.{WithDiscoveryFlinkSQLEngine, WithFlinkSQLEngineOnYarn} -import org.apache.kyuubi.engine.flink.FlinkEngineUtils.FLINK_RUNTIME_VERSION import org.apache.kyuubi.ha.HighAvailabilityConf.{HA_ENGINE_REF_ID, HA_NAMESPACE} class FlinkOperationOnYarnSuite extends FlinkOperationSuite with WithDiscoveryFlinkSQLEngine with WithFlinkSQLEngineOnYarn { - protected def jdbcUrl: String = { - // FLINK-38974 (2.3.0) rejects jobs that are not associated with an application registered in - // the dispatcher. The engine submits jobs through its own copy of EmbeddedExecutorFactory, - // which does not stamp an application id, so every query fails on Flink 2.3 and later. - assume( - FLINK_RUNTIME_VERSION < "2.3", - "Flink application mode is not supported since Flink 2.3, see FLINK-38974") - getFlinkEngineServiceUrl - } + protected def jdbcUrl: String = getFlinkEngineServiceUrl override def withKyuubiConf: Map[String, String] = { Map( diff --git a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala index 6c8681b8cce..c06b742d7aa 100644 --- a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala +++ b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuiteOnYarn.scala @@ -17,25 +17,17 @@ package org.apache.kyuubi.it.flink.operation -import org.apache.flink.runtime.util.EnvironmentInformation - import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf._ import org.apache.kyuubi.it.flink.WithKyuubiServerAndYarnMiniCluster import org.apache.kyuubi.operation.HiveJDBCTestHelper import org.apache.kyuubi.operation.meta.ResultSetSchemaConstant.TABLE_CAT import org.apache.kyuubi.shaded.hive.service.rpc.thrift.{TGetInfoReq, TGetInfoType} -import org.apache.kyuubi.util.SemanticVersion class FlinkOperationSuiteOnYarn extends WithKyuubiServerAndYarnMiniCluster with HiveJDBCTestHelper { override protected def jdbcUrl: String = { - // FLINK-38974 (2.3.0) rejects jobs that are not associated with an application registered in - // the dispatcher, so application mode does not work on Flink 2.3 and later - assume( - SemanticVersion(EnvironmentInformation.getVersion) < "2.3", - "Flink application mode is not supported since Flink 2.3, see FLINK-38974") // delay the access to thrift service because the thrift service // may not be ready although it's registered Thread.sleep(3000L)