[FLINK-40403][runtime] Pre-compile subtask-index regexes in MetricStore

[NeQuissimus]

What is the purpose of the change

MetricStore.TaskMetricStore#retainSubtasks and #isTransientMetric identify the
per-subtask key prefix (e.g. "0.numRecordsIn") with String#matches(String), which
recompiles the regex Pattern on every call. Both run once per metric key, and
retainSubtasks is invoked once per vertex on every MetricFetcher refresh from
MetricStore#updateCurrentExecutionAttempts, which is synchronized — so the
O(#metric-keys) work, including a fresh Pattern.compile per key, executes while the
single global MetricStore monitor is held.

On jobs with a large number of subtasks this per-refresh compilation dominates the time
the monitor is held. Because the JobManager REST handler thread pool
(rest.server.numThreads) is shared between the MetricFetcher and the REST endpoints,
the whole pool stalls on that monitor: GET /jobs/<jid> (which aggregates per-subtask IO
metrics via MutableIOMetrics#addIOMetrics, taking the same monitor once per subtask)
becomes very slow, and endpoints that never touch MetricStore (/jobs/overview,
/jobs/<jid>/checkpoints) queue behind the saturated pool. Observed on a ~2,233-subtask
job (main chain at parallelism 900): a thread dump taken while GET /jobs/<jid> hung
showed the entire REST pool contending on the MetricStore monitor, one thread RUNNABLE
inside java.util.regex.Pattern/String#matches while holding it. CPU was otherwise
near-idle — this is lock-hold time, not compute.

Brief change log

• Add two static final Pattern fields to MetricStore.TaskMetricStore and match via
  matcher(...).matches() in retainSubtasks and isTransientMetric, instead of
  String#matches(String) which recompiles the pattern on every call. The regex strings
  are unchanged.

Verifying this change

This change is a behaviour-preserving refactor already covered by existing tests:
MetricStoreTest#testTaskMetricStoreCleanup (drives retainSubtasks via
updateCurrentExecutionAttempts), #testSubtaskMetricStoreCleanup, and
#testMalformedNameHandling (empty/null metric names). String#matches(regex) is
specified as Pattern.compile(regex).matcher(s).matches(), so pre-compiling the identical
patterns does not change matching behaviour.

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 (REST/metrics query path only)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

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

• Yes (please specify the tool below)

Generated-by: Pi (Anthropic claude-opus-4-8)

[flinkbot]

CI report:

• e6ec59c Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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