[FLINK-36755][k8s] Relax the limitation that kubernetes application require a main jar

Author: fsk119

flink-clients/src/main/java/org/apache/flink/client/program/artifact/ArtifactFetchManager.java

@@ -115,11 +115,9 @@ public Result fetchArtifacts(String[] uris) {
      * @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()

flink-clients/src/test/java/org/apache/flink/client/program/artifact/ArtifactFetchManagerTest.java

@@ -198,9 +198,6 @@ void testHttpDisabledError() {
     @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)

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java

@@ -220,7 +220,8 @@ public ClusterClientProvider<String> deployApplicationCluster(
                 || 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 {

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 static void main(final String[] args) {
         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 @@ private static ArtifactFetchManager.Result fetchArtifacts(Configuration configur
             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);
         }

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java

@@ -400,7 +400,8 @@ public static List<String> getStartCommandWithBashWrapper(String command) {
     }
 
     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());
     }

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java

@@ -157,6 +157,21 @@ void testDeployApplicationCluster() {
         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 @@ void testDeployApplicationClusterWithMultipleJarsSet() {
                         cause ->
                                 assertThat(cause)
                                         .isInstanceOf(IllegalArgumentException.class)
-                                        .hasMessageContaining("Should only have one jar"));
+                                        .hasMessageContaining("Should only have at most one jar"));
     }
 
     @Test

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 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 void testGenerateJarDir() {
                         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();
+    }
 }