[FLINK-21210][coordination] ApplicationClusterEntryPoint explicitly c…

…loses PackagedProgram

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

@@ -37,6 +37,7 @@
 import org.apache.flink.runtime.resourcemanager.ResourceManagerFactory;
 import org.apache.flink.runtime.rest.JobRestEndpointFactory;
 
+import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.Collections;
@@ -47,7 +48,7 @@
 
 /**
  * Base class for cluster entry points targeting executing applications in "Application Mode". The
- * lifecycle of the enrtypoint is bound to that of the specific application being executed, and the
+ * lifecycle of the entry point is bound to that of the specific application being executed, and the
  * {@code main()} method of the application is run on the cluster.
  */
 public class ApplicationClusterEntryPoint extends ClusterEntrypoint {
@@ -114,4 +115,11 @@ private static List<URL> getClasspath(
  return Collections.unmodifiableList(
  classpath.stream().distinct().collect(Collectors.toList()));
  }
+
+ @Override
+ protected void cleanupDirectories() throws IOException {
+ // Close the packaged program explicitly to clean up temporary jars.
+ program.close();
+ super.cleanupDirectories();
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java

@@ -506,7 +506,7 @@ private CompletableFuture<Void> closeClusterComponent(
  *
  * @throws IOException if the temporary directories could not be cleaned up
  */
- private void cleanupDirectories() throws IOException {
+ protected void cleanupDirectories() throws IOException {
  ShutdownHookUtil.removeShutdownHook(shutDownHook, getClass().getSimpleName(), LOG);
 
  final String webTmpDir = configuration.getString(WebOptions.TMP_DIR);