[FLINK-40294][table-planner] Instrument async scalar and table UDF calls with metrics

[weiqingy]

This is the third PR of the FLIP-485 implementation, split into a stack of small, independently reviewable PRs under the umbrella issue FLINK-38071. Landing order:

Step	Sub-task	Scope
PR-1 ✅ merged	FLINK-40292	UdfMetrics helper: registration, sampling, timing, exception counting (#28878)
PR-2 🔵 in review	FLINK-40293	Config options + sync scalar/table instrumentation (#28879)
PR-3 (this PR)	FLINK-40294	Async scalar/table instrumentation
PR-4	FLINK-40295	User-facing documentation

Stacked on #28879 and kept as a draft until that merges, so the diff here currently also shows PR-2's commit. It will collapse to this step's changes once PR-2 lands.

What is the purpose of the change

Extends the FLIP-485 UDF metrics to asynchronous scalar and table functions, so udfProcessingTime and udfExceptionCount cover the async paths as well as the synchronous ones.

An async invocation does not begin and end on the same thread, so the instrumentation is split. At dispatch, on the task thread, the sampling decision is taken and the start timestamp captured. At completion, on the callback thread, the elapsed time is recorded and an exceptional completion increments the counter. udfProcessingTime therefore measures the full dispatch-to-completion span, not just the synchronous hand-off.

Brief change log

• Emit the dispatch-side instrumentation for the async scalar and async table call sites in AsyncCodeGenerator and BridgingFunctionGenUtil.
• Record completion in the per-invocation delegating futures, DelegatingAsyncResultFuture (async scalar) and DelegatingAsyncTableResultFuture (async table).
• Add a single-handle accessor to CodeGeneratorContext so the generated fetcher resolves the shared UdfMetrics handle.
• Write the sample decision and start timestamp before registering the completion callback, establishing a happens-before edge to the callback thread. The histogram is synchronized and the exception counter is thread-safe.

As on the synchronous path, the instrumentation is emitted at code generation only when table.exec.udf-metric-enabled is true, so the generated code is unchanged when the feature is disabled.

Known limitation: an async scalar call whose arguments are all null short-circuits but still flows through the future, so with metrics enabled it records one near-zero histogram entry. This is a bounded skew in the timing distribution, not a correctness issue; avoiding it cleanly needs a separate dispatch path for the short-circuit case.

Verifying this change

This change added tests and can be verified as follows:

• New async cases in UdfMetricsITCase: async scalar and async table metrics, naming and scope, and an exceptional completion that increments udfExceptionCount while the job still finishes.
• The existing AsyncCalcITCase and AsyncCorrelateITCase suites guard the async paths against regressions.

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): yes, the async dispatch and completion paths. Gated at code generation, so the generated code is unchanged when the feature is disabled.
• 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? yes
• If yes, how is the feature documented? JavaDocs here; the metrics.md section lands in PR-4.

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Anthropic Claude Opus 4.8 and Claude Opus 5)

[flinkbot]

CI report:

• 0440e15 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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