[FLINK-38758][runtime] Introduce PackagedProgramApplication

Author: eemario

docs/layouts/shortcodes/generated/deployment_configuration.html

@@ -62,5 +62,11 @@
             <td>String</td>
             <td>The deployment target for the execution. This can take one of the following values when calling <code class="highlighter-rouge">bin/flink run</code>:<ul><li>remote</li><li>local</li><li>yarn-application</li><li>yarn-session</li><li>kubernetes-application</li><li>kubernetes-session</li></ul></td>
         </tr>
+        <tr>
+            <td><h5>execution.terminate-application-on-any-job-terminated-exceptionally</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>When it is set to true, the application will complete exceptionally if any job fails or is canceled. When it is set to false, the application will finish after all jobs reach terminal states.</td>
+        </tr>
     </tbody>
 </table>

flink-clients/src/main/java/org/apache/flink/client/deployment/application/PackagedProgramApplication.java

@@ -98,6 +98,9 @@ public class PackagedProgram implements AutoCloseable {
     /** Flag indicating whether the job is a Python job. */
     private final boolean isPython;
 
+    /** Serializable descriptor to reconstruct the PackagedProgram. */
+    private final PackagedProgramDescriptor descriptor;
+
     /**
      * Creates an instance that wraps the plan defined in the jar file using the given arguments.
      * For generating the plan the class defined in the className parameter is used.
@@ -163,6 +166,19 @@ private PackagedProgram(
             throw new ProgramInvocationException(
                     "The given program class does not have a main(String[]) method.");
         }
+
+        this.descriptor =
+                new PackagedProgramDescriptor(
+                        jarFile,
+                        classpaths,
+                        configuration,
+                        savepointRestoreSettings,
+                        args,
+                        getMainClassName());
+    }
+
+    public PackagedProgramDescriptor getDescriptor() {
+        return descriptor;
     }
 
     public SavepointRestoreSettings getSavepointSettings() {

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

@@ -0,0 +1,101 @@
+/*
+ * 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.client.program;
+
+import org.apache.flink.client.deployment.application.PackagedProgramApplication;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.Serializable;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Descriptor for {@link PackagedProgram}.
+ *
+ * <p>This class provides a serializable representation of {@link PackagedProgram} that can be used
+ * to reconstruct the {@link PackagedProgram} after serialization and deserialization. It is mainly
+ * used by {@link PackagedProgramApplication}.
+ */
+public class PackagedProgramDescriptor implements Serializable {
+
+    @Nullable private final File jarFile;
+
+    private final List<URL> userClassPaths;
+
+    private final Configuration configuration;
+
+    private final SavepointRestoreSettings savepointRestoreSettings;
+
+    private final String[] programArgs;
+
+    private final String mainClassName;
+
+    public PackagedProgramDescriptor(
+            @Nullable File jarFile,
+            List<URL> userClassPaths,
+            Configuration configuration,
+            SavepointRestoreSettings savepointRestoreSettings,
+            String[] programArgs,
+            String mainClassName) {
+        this.jarFile = jarFile;
+        this.userClassPaths = userClassPaths;
+        this.configuration = configuration;
+        this.savepointRestoreSettings = savepointRestoreSettings;
+        this.programArgs = programArgs;
+        this.mainClassName = mainClassName;
+    }
+
+    public String getMainClassName() {
+        return mainClassName;
+    }
+
+    public PackagedProgram toPackageProgram() throws ProgramInvocationException {
+        return PackagedProgram.newBuilder()
+                .setJarFile(jarFile)
+                .setEntryPointClassName(mainClassName)
+                .setConfiguration(configuration)
+                .setUserClassPaths(userClassPaths)
+                .setArguments(programArgs)
+                .setSavepointRestoreSettings(savepointRestoreSettings)
+                .build();
+    }
+
+    @Override
+    public String toString() {
+        return "PackagedProgramDescriptor{"
+                + "jarFile="
+                + jarFile
+                + ", userClassPaths="
+                + userClassPaths
+                + ", configuration="
+                + configuration
+                + ", savepointRestoreSettings="
+                + savepointRestoreSettings
+                + ", programArgs="
+                + Arrays.toString(programArgs)
+                + ", mainClassName="
+                + mainClassName
+                + '}';
+    }
+}