[Bug]: Flink Job manager metaspace classloader leak caused by not releasing PipelineOptionFactory DESERIALIZATION_CONTEXT

[jinyangzhen]

What happened?

After a beam job (using Flink runner) submitted, the Flink ChildFirstClassLoader is not released after job completion. Observing from its GCRoot, this classloader is not released due to a thread local static variable DESERIALIZATION_CONTEXT being hold by a parking Flink thread. For example:

Here is a small example to reproduce this issue.
word-count-beam.tar.gz
Notice: please copy the source file kinglear.txt to /tmp before running this example

I tested this example against below 2 envs locally:
Flink 1.13 + beam-flink-runner-1.13, both JM and TM can observe the same leak.
Flink 1.16 + beam-flink-runner-1.16, only JM can observe this leak.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[jinyangzhen]

the same issue as #25510

and this bug was introduced from this PR #16680