You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fc...@apache.org on 2025/07/03 05:18:09 UTC
(flink) branch release-2.0 updated: [FLINK-33926][k8s] Allow using job JARs in the system classpath in native mode
This is an automated email from the ASF dual-hosted git repository.
fcsaky pushed a commit to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.0 by this push:
new 0a278ed2ddc [FLINK-33926][k8s] Allow using job JARs in the system classpath in native mode
0a278ed2ddc is described below
commit 0a278ed2ddcf03921382db889d9e7b7b279dcc8c
Author: Ferenc Csaky <fc...@apache.org>
AuthorDate: Thu Jul 3 07:18:03 2025 +0200
[FLINK-33926][k8s] Allow using job JARs in the system classpath in native mode
Co-authored-by: Gantigmaa Selenge <39...@users.noreply.github.com>
---
.../org/apache/flink/client/program/PackagedProgramUtils.java | 5 +++++
.../org/apache/flink/kubernetes/KubernetesClusterDescriptor.java | 6 ++++--
.../entrypoint/KubernetesApplicationClusterEntrypoint.java | 5 +++--
.../apache/flink/kubernetes/KubernetesClusterDescriptorTest.java | 9 +++++++++
4 files changed, 21 insertions(+), 4 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
index 7833b179bc0..9ea781ddc37 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.dag.Pipeline;
import org.apache.flink.client.FlinkPipelineTranslationUtil;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.commons.collections.CollectionUtils;
@@ -219,6 +220,10 @@ public enum PackagedProgramUtils {
return new File(path).getAbsoluteFile().toURI();
}
+ public static boolean usingSystemClassPath(Configuration configuration) {
+ return configuration.get(PipelineOptions.JARS) == null;
+ }
+
private static ProgramInvocationException generateException(
PackagedProgram program,
String msg,
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
index 865dd7d219c..009f6b6341b 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
@@ -214,9 +214,11 @@ public class KubernetesClusterDescriptor implements ClusterDescriptor<String> {
applicationConfiguration.applyToConfiguration(flinkConfig);
- // No need to do pipelineJars validation if it is a PyFlink job.
+ // No need to do pipelineJars validation if it is a PyFlink job or using system classpath.
if (!(PackagedProgramUtils.isPython(applicationConfiguration.getApplicationClassName())
- || PackagedProgramUtils.isPython(applicationConfiguration.getProgramArguments()))) {
+ || PackagedProgramUtils.isPython(
+ applicationConfiguration.getProgramArguments()))
+ && !PackagedProgramUtils.usingSystemClassPath(flinkConfig)) {
final List<URI> pipelineJars =
KubernetesUtils.checkJarFileForApplicationMode(flinkConfig);
Preconditions.checkArgument(
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypoint.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypoint.java
index 37499a39c6c..41e14e1c91c 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypoint.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypoint.java
@@ -125,9 +125,10 @@ public final class KubernetesApplicationClusterEntrypoint extends ApplicationClu
final File userLibDir = ClusterEntrypointUtils.tryFindUserLibDirectory().orElse(null);
- // No need to do pipelineJars validation if it is a PyFlink job.
+ // No need to do pipelineJars validation if it is a PyFlink job or using system classpath
if (!(PackagedProgramUtils.isPython(jobClassName)
- || PackagedProgramUtils.isPython(programArguments))) {
+ || PackagedProgramUtils.isPython(programArguments))
+ && !PackagedProgramUtils.usingSystemClassPath(configuration)) {
final ArtifactFetchManager.Result fetchRes = fetchArtifacts(configuration);
return DefaultPackagedProgramRetriever.create(
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
index 93ef49a497c..5dda11ed4f2 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
@@ -52,6 +52,7 @@ import java.util.concurrent.Executors;
import static org.apache.flink.kubernetes.utils.Constants.ENV_FLINK_POD_IP_ADDRESS;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link KubernetesClusterDescriptor}. */
@@ -220,6 +221,14 @@ class KubernetesClusterDescriptorTest extends KubernetesClientTestBase {
.hasMessageContaining("Should only have at most one jar"));
}
+ @Test
+ void testDeployApplicationClusterWithNoJar() {
+ flinkConfig.set(DeploymentOptions.TARGET, KubernetesDeploymentTarget.APPLICATION.getName());
+ assertThatNoException()
+ .isThrownBy(
+ () -> descriptor.deployApplicationCluster(clusterSpecification, appConfig));
+ }
+
@Test
void testDeployApplicationClusterWithClusterIP() throws Exception {
flinkConfig.set(