From 26e5e93e1798c1c1c25c92b346c0922efc751c94 Mon Sep 17 00:00:00 2001 From: huanxiaodong Date: Wed, 26 Aug 2026 17:41:24 +0800 Subject: [PATCH] [FLINK-40473] Fix yarn-application mode failing to locate flink-cdc-dist jar with Flink version suffix Signed-off-by: huanxiaodong --- .../YarnApplicationDeploymentExecutor.java | 21 +++++- ...YarnApplicationDeploymentExecutorTest.java | 73 +++++++++++++++++++ 2 files changed, 90 insertions(+), 4 deletions(-) create mode 100644 flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutorTest.java diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutor.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutor.java index c8ff52253a0..ece4c069869 100644 --- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutor.java +++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutor.java @@ -17,6 +17,7 @@ package org.apache.flink.cdc.composer.flink.deployment; +import org.apache.flink.cdc.common.annotation.VisibleForTesting; import org.apache.flink.cdc.common.utils.Preconditions; import org.apache.flink.cdc.composer.PipelineDeploymentExecutor; import org.apache.flink.cdc.composer.PipelineExecution; @@ -59,8 +60,8 @@ public class YarnApplicationDeploymentExecutor implements PipelineDeploymentExec LoggerFactory.getLogger(YarnApplicationDeploymentExecutor.class); private static final String FLINK_CDC_HOME_ENV_VAR = "FLINK_CDC_HOME"; - private static final String FLINK_CDC_DIST_JAR_PATTERN = - "^flink-cdc-dist-(\\d+(\\.\\d+)*)(-SNAPSHOT)?\\.jar$"; + private static final String FLINK_CDC_DIST_JAR_PREFIX = "flink-cdc-dist-"; + private static final String JAR_SUFFIX = ".jar"; private static final String APPLICATION_MAIN_CLASS = "org.apache.flink.cdc.cli.CliExecutor"; @Override @@ -128,7 +129,16 @@ private String getFlinkCDCDistJarFromEnv() throws IOException { flinkCDCHomeFromEnvVar, "FLINK_CDC_HOME is not correctly set in environment variable, current FLINK_CDC_HOME is: " + flinkCDCHomeFromEnvVar); - Path flinkCDCLibPath = new Path(flinkCDCHomeFromEnvVar, "lib"); + return getFlinkCDCDistJar(new Path(flinkCDCHomeFromEnvVar, "lib")); + } + + /** + * Finds the Flink CDC dist jar under the given lib directory. The jar is matched by name prefix + * instead of a version pattern, because binaries released since 3.6.0 are built once per Flink + * minor version and carry an extra suffix, e.g. {@code flink-cdc-dist-3.6.0-1.20.jar}. + */ + @VisibleForTesting + static String getFlinkCDCDistJar(Path flinkCDCLibPath) throws IOException { if (!flinkCDCLibPath.getFileSystem().exists(flinkCDCLibPath) || !flinkCDCLibPath.getFileSystem().getFileStatus(flinkCDCLibPath).isDir()) { throw new RuntimeException( @@ -141,7 +151,10 @@ private String getFlinkCDCDistJarFromEnv() throws IOException { Arrays.stream(fileStatuses) .filter(status -> !status.isDir()) .map(FileStatus::getPath) - .filter(path -> path.getName().matches(FLINK_CDC_DIST_JAR_PATTERN)) + .filter( + path -> + path.getName().startsWith(FLINK_CDC_DIST_JAR_PREFIX) + && path.getName().endsWith(JAR_SUFFIX)) .findFirst(); if (distJars.isPresent()) { diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutorTest.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutorTest.java new file mode 100644 index 00000000000..aa49991fc23 --- /dev/null +++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutorTest.java @@ -0,0 +1,73 @@ +/* + * 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.flink.cdc.composer.flink.deployment; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.FileNotFoundException; +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test for {@link YarnApplicationDeploymentExecutor}. */ +class YarnApplicationDeploymentExecutorTest { + + private static final String CONNECTOR_JAR = "flink-cdc-pipeline-connector-mysql-3.6.0.jar"; + + @ParameterizedTest + @ValueSource( + strings = { + "flink-cdc-dist-3.5.0.jar", + "flink-cdc-dist-3.7-SNAPSHOT.jar", + // Binaries released since 3.6.0 carry a Flink version suffix. + "flink-cdc-dist-3.6.0-1.20.jar", + "flink-cdc-dist-3.7.0-2.0.jar" + }) + void testFindDistJarWhateverVersionItCarries(String distJarName, @TempDir Path libDir) + throws Exception { + Files.createFile(libDir.resolve(distJarName)); + // Connector jars share the same directory and must not be picked up. + Files.createFile(libDir.resolve(CONNECTOR_JAR)); + // A directory named like the dist jar must not be picked up either. + Files.createDirectory(libDir.resolve("flink-cdc-dist-directory.jar")); + + assertThat(YarnApplicationDeploymentExecutor.getFlinkCDCDistJar(toFlinkPath(libDir))) + .endsWith(distJarName); + } + + @Test + void testDistJarNotFound(@TempDir Path libDir) throws Exception { + Files.createFile(libDir.resolve(CONNECTOR_JAR)); + + assertThatThrownBy( + () -> + YarnApplicationDeploymentExecutor.getFlinkCDCDistJar( + toFlinkPath(libDir))) + .isInstanceOf(FileNotFoundException.class) + .hasMessageContaining("Failed to fetch Flink CDC dist jar"); + } + + private static org.apache.flink.core.fs.Path toFlinkPath(Path path) { + return new org.apache.flink.core.fs.Path(path.toUri()); + } +}