[FLINK-35376][runtime] Avoid duplicate PackagedProgram in REST JAR handlers

[jagath25]

What is the purpose of the change

REST JAR handlers currently build a PackagedProgram in JarHandlerContext.applyToConfiguration and build another one for run or plan execution. The first program is never closed. Its user-code classloader can keep the uploaded JAR open after the upload is deleted, causing deleted files to continue consuming JobManager disk space on repeated submissions. This addresses FLINK-35376.

Brief change log

• Apply request options before constructing a program, then derive JAR and classpath options from the program owned by the run, plan, or application handler.
• Close the run or application program if it fails before ownership is handed off, and close the application program only before the PackagedProgramApplication is created. Afterwards the application owns the program, and Dispatcher.submitApplication disposes applications it rejects or fails to register.
• Close the PackagedProgram held by PackagedProgramApplication when the application finishes or is disposed.
• JarPlanHandler now builds its program inside the async task, so a missing JAR returns 400 instead of an unstripped CompletionException (500).
• Add tests showing that applying request options does not open the JAR, that program-derived JARS/CLASSPATHS are set, that extracted nested libraries live until the program is closed, that PackagedProgramApplication closes its program on finish, on dispose and after recovery, that a rejected duplicate application is disposed, and that a missing JAR fails the plan handler asynchronously with 400.

Verifying this change

• JarHandlerUtilsTest, JarRunHandlerParameterTest, JarPlanHandlerParameterTest, and JarRunApplicationHandlerParameterTest: 38 tests (plus PackagedProgramApplicationTest, 37 tests, and DispatcherApplicationTest, 27 tests) passed with mvn -pl flink-runtime-web -am ... test on JDK 17; Spotless and git diff --check passed.
• Manually exercised the Flink 1.20.1 REST upload/run/delete path in an isolated session cluster: before the change, six finished WordCount jobs left six deleted-open upload JARs; with the change, the same three batch and three streaming submissions left none.
• Also manually verified on a master (2.4-SNAPSHOT) build of this branch, covering both REST call sites in a local session cluster on JDK 17:
  • POST /jars/upload -> POST /jars/:jarId/run -> wait FINISHED -> DELETE /jars/:jarId, three times: lsof on the JobManager showed no deleted-but-open upload JARs.
  • POST /jars/:jarId/run-application: HTTP 202, job FINISHED; a second submission with the same applicationId was rejected with HTTP 400 (DuplicateApplicationSubmissionException) and disposed; after deleting both uploaded JARs, lsof again showed no deleted-but-open JARs.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager REST submission, yes; checkpointing and recovery, no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: OpenAI Codex (GPT-6)

[flinkbot]

CI report:

• a5c7f9d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[jagath25]

@flinkbot run azure

[jagath25]

@flinkbot run azure