You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2024/11/25 11:11:53 UTC
(flink) 01/02: [FLINK-36755][k8s] Relax the limitation that kubernetes application require a main jar
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0f1367a0fa12c18a3d559f43dc9a32307eaad444
Author: Shengkai <10...@qq.com>
AuthorDate: Fri Nov 22 11:24:02 2024 +0800
[FLINK-36755][k8s] Relax the limitation that kubernetes application require a main jar
---
.../client/program/artifact/ArtifactFetchManager.java | 6 ++----
.../program/artifact/ArtifactFetchManagerTest.java | 3 ---
.../flink/kubernetes/KubernetesClusterDescriptor.java | 3 ++-
.../KubernetesApplicationClusterEntrypoint.java | 11 +++++++----
.../apache/flink/kubernetes/utils/KubernetesUtils.java | 3 ++-
.../kubernetes/KubernetesClusterDescriptorTest.java | 17 ++++++++++++++++-
.../KubernetesApplicationClusterEntrypointTest.java | 15 +++++++++++++++
7 files changed, 44 insertions(+), 14 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/artifact/ArtifactFetchManager.java b/flink-clients/src/main/java/org/apache/flink/client/program/artifact/ArtifactFetchManager.java
index f37d303e12f..caf9664535c 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/artifact/ArtifactFetchManager.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/artifact/ArtifactFetchManager.java
@@ -115,11 +115,9 @@ public class ArtifactFetchManager {
* @return result with the fetched artifacts
* @throws Exception
*/
- public Result fetchArtifacts(String jobUri, @Nullable List<String> additionalUris)
+ public Result fetchArtifacts(@Nullable String jobUri, @Nullable List<String> additionalUris)
throws Exception {
- checkArgument(jobUri != null && !jobUri.trim().isEmpty(), "The jobUri is required.");
-
- File jobJar = fetchArtifact(jobUri);
+ File jobJar = jobUri == null ? null : fetchArtifact(jobUri);
List<File> additionalArtifacts =
additionalUris == null
? Collections.emptyList()
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/artifact/ArtifactFetchManagerTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/artifact/ArtifactFetchManagerTest.java
index d27ab8deac5..cb36f76450a 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/artifact/ArtifactFetchManagerTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/artifact/ArtifactFetchManagerTest.java
@@ -198,9 +198,6 @@ class ArtifactFetchManagerTest {
@Test
void testMissingRequiredFetchArgs() {
ArtifactFetchManager fetchMgr = new ArtifactFetchManager(configuration);
- assertThatThrownBy(() -> fetchMgr.fetchArtifacts(null, null))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessage("The jobUri is required.");
assertThatThrownBy(() -> fetchMgr.fetchArtifacts(null))
.isInstanceOf(IllegalArgumentException.class)
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 b81520b55e5..25547a1193d 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
@@ -220,7 +220,8 @@ public class KubernetesClusterDescriptor implements ClusterDescriptor<String> {
|| PackagedProgramUtils.isPython(applicationConfiguration.getProgramArguments()))) {
final List<URI> pipelineJars =
KubernetesUtils.checkJarFileForApplicationMode(flinkConfig);
- Preconditions.checkArgument(pipelineJars.size() == 1, "Should only have one jar");
+ Preconditions.checkArgument(
+ pipelineJars.size() <= 1, "Should only have at most one jar.");
}
try {
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 e3c7672122c..37499a39c6c 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
@@ -19,6 +19,7 @@
package org.apache.flink.kubernetes.entrypoint;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.client.cli.ArtifactFetchOptions;
import org.apache.flink.client.deployment.application.ApplicationClusterEntryPoint;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
@@ -101,7 +102,8 @@ public final class KubernetesApplicationClusterEntrypoint extends ApplicationClu
ClusterEntrypoint.runClusterEntrypoint(kubernetesApplicationClusterEntrypoint);
}
- private static PackagedProgram getPackagedProgram(final Configuration configuration)
+ @VisibleForTesting
+ static PackagedProgram getPackagedProgram(final Configuration configuration)
throws FlinkException {
final ApplicationConfiguration applicationConfiguration =
@@ -146,14 +148,15 @@ public final class KubernetesApplicationClusterEntrypoint extends ApplicationClu
String targetDir = generateJarDir(configuration);
ArtifactFetchManager fetchMgr = new ArtifactFetchManager(configuration, targetDir);
- List<String> uris = configuration.get(PipelineOptions.JARS);
- checkArgument(uris.size() == 1, "Should only have one jar");
+ List<String> uris =
+ configuration.getOptional(PipelineOptions.JARS).orElse(Collections.emptyList());
+ checkArgument(uris.size() <= 1, "Should only have at most one jar");
List<String> additionalUris =
configuration
.getOptional(ArtifactFetchOptions.ARTIFACT_LIST)
.orElse(Collections.emptyList());
- return fetchMgr.fetchArtifacts(uris.get(0), additionalUris);
+ return fetchMgr.fetchArtifacts(uris.size() == 1 ? uris.get(0) : null, additionalUris);
} catch (Exception e) {
throw new RuntimeException(e);
}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java
index bc2b65920ad..d503ffff649 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java
@@ -400,7 +400,8 @@ public class KubernetesUtils {
}
public static List<URI> checkJarFileForApplicationMode(Configuration configuration) {
- return configuration.get(PipelineOptions.JARS).stream()
+ return configuration.getOptional(PipelineOptions.JARS).orElse(Collections.emptyList())
+ .stream()
.map(FunctionUtils.uncheckedFunction(PackagedProgramUtils::resolveURI))
.collect(Collectors.toList());
}
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 a08263e988c..93ef49a497c 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
@@ -157,6 +157,21 @@ class KubernetesClusterDescriptorTest extends KubernetesClientTestBase {
checkUpdatedConfigAndResourceSetting();
}
+ @Test
+ void testDeployApplicationClusterWithoutJar() {
+ flinkConfig.set(DeploymentOptions.TARGET, KubernetesDeploymentTarget.APPLICATION.getName());
+ try {
+ descriptor.deployApplicationCluster(clusterSpecification, appConfig);
+ } catch (Exception ignored) {
+ }
+
+ mockExpectedServiceFromServerSide(loadBalancerSvc);
+ final ClusterClient<String> clusterClient =
+ descriptor.retrieve(CLUSTER_ID).getClusterClient();
+ checkClusterClient(clusterClient);
+ checkUpdatedConfigAndResourceSetting();
+ }
+
@Test
void testDeployApplicationClusterWithClusterAlreadyExists() {
flinkConfig.set(
@@ -202,7 +217,7 @@ class KubernetesClusterDescriptorTest extends KubernetesClientTestBase {
cause ->
assertThat(cause)
.isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("Should only have one jar"));
+ .hasMessageContaining("Should only have at most one jar"));
}
@Test
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypointTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypointTest.java
index ac3ef5c6fdc..4b6e02459f8 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypointTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/entrypoint/KubernetesApplicationClusterEntrypointTest.java
@@ -19,6 +19,8 @@
package org.apache.flink.kubernetes.entrypoint;
import org.apache.flink.client.cli.ArtifactFetchOptions;
+import org.apache.flink.client.deployment.application.ApplicationConfiguration;
+import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
@@ -32,6 +34,8 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.nio.file.Path;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Tests for {@link KubernetesApplicationClusterEntrypointTest}. */
public class KubernetesApplicationClusterEntrypointTest {
private static final Logger LOG =
@@ -58,4 +62,15 @@ public class KubernetesApplicationClusterEntrypointTest {
new String[] {tempDir.toString(), TEST_NAMESPACE, TEST_CLUSTER_ID});
Assertions.assertEquals(expectedDir, baseDir);
}
+
+ @Test
+ void testGetPackagedProgram() throws Exception {
+ Configuration config = new Configuration();
+ new ApplicationConfiguration(
+ new String[0], KubernetesApplicationClusterEntrypoint.class.getName())
+ .applyToConfiguration(config);
+ PackagedProgram packagedProgram =
+ KubernetesApplicationClusterEntrypoint.getPackagedProgram(config);
+ assertThat(packagedProgram.getJobJarAndDependencies()).isNullOrEmpty();
+ }
}