[FLINK-40292][table-runtime] Add UdfMetrics helper for UDF metrics

[weiqingy]

This is the first 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 (this PR)	FLINK-40292	UdfMetrics helper: registration, sampling, timing, exception counting
PR-2	FLINK-40293	Config options + sync scalar/table instrumentation
PR-3	FLINK-40294	Async scalar/table instrumentation
PR-4	FLINK-40295	User-facing documentation

FLIP-485 passed the vote on 2026-08-01. This series supersedes the single reference-implementation draft in #28692, which is closed.

What is the purpose of the change

FLIP-485 adds opt-in, per-operator observability for SQL/Table user-defined functions, so operators can see inside UDF "black boxes" when debugging latency or errors, and so autoscaling gets a reliable "the problem is in user code" signal.

This PR adds only the shared runtime helper, UdfMetrics. It owns metric registration, the sampling decision, timing, and exception counting, and it is the piece both the synchronous and the asynchronous paths use. It has no caller yet; the first caller arrives in PR-2. Keeping it separate makes the sampling and registration logic reviewable on its own.

Brief change log

• Add UdfMetrics in flink-table-runtime (org.apache.flink.table.runtime.operators.metrics, beside SimpleGauge). It registers udfProcessingTime and udfExceptionCount under addGroup("udf", udfName) on the operator metric group, so the full identifier is <operator_name>.udf.<udf_name>.<metric>.
• udfProcessingTime is a DescriptiveStatisticsHistogram of per-invocation nanoseconds, sampled with the same counter-based scheme as state latency tracking (FLINK-21736), including the interval == 1 "measure every call" case.
• udfExceptionCount is a ThreadSafeSimpleCounter, incremented on every exception and not sampled.
• The histogram is safe to update from an async callback thread; the sampling counter is only advanced on the task thread at dispatch.

Verifying this change

This change added tests and can be verified as follows:

• UdfMetricsTest covers metric registration and naming, the sampling decision across the interval boundary, the sample-interval = 1 case, rejection of a non-positive interval, timing recorded into the histogram, exception counting, and that two UDFs registered on the same operator get independent metrics.

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. Nothing calls this class yet; the call sites and their gating arrive in PR-2 and PR-3.
• 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, as the first step of FLIP-485
• If yes, how is the feature documented? JavaDocs here; the user-facing documentation 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)

[weiqingy]

This is the first of four PRs landing the accepted design (FLIP-485). It replaces the single reference-implementation draft in #28692, which I'm closing.

This PR adds only the UdfMetrics helper and its unit test. It has no caller yet on purpose. The first caller arrives in PR 2, which adds the two config options and instruments synchronous scalar and table UDF calls. Splitting it out keeps the sampling and registration logic reviewable on its own, since it is the piece both the sync and async paths share.

The series:

1. UdfMetrics helper (this PR)
2. table.exec.udf-metric-enabled / table.exec.udf-metric.sample-interval plus synchronous scalar and table instrumentation
3. Asynchronous scalar and table instrumentation
4. Documentation

Each PR builds and passes its own tests independently. I'll open the next one as the previous merges.

Hi @xuyangzhong @RocMarshal @HuangZhenQiu, since you voted on the FLIP, I’d appreciate it if you could take a look when you have time.

[flinkbot]

CI report:

• 15bcbfa Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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