weiqingy opened a new pull request, #28692: URL: https://github.com/apache/flink/pull/28692
## What is the purpose of the change This pull request is a **reference implementation of [FLIP-485: Add UDF Metrics](https://cwiki.apache.org/confluence/spaces/FLINK/pages/373885706/FLIP-485+Add+UDF+Metrics)** ([FLINK-38071](https://issues.apache.org/jira/browse/FLINK-38071)). It adds opt-in, per-operator observability for SQL/Table user-defined functions so operators can see inside UDF "black boxes" when debugging latency, throughput, or errors — and feed a reliable "the problem is in user code" signal to autoscaling. Two metrics are registered on the executing operator's `OperatorMetricGroup`, scoped as `<operator_name>.udf.<udf_name>.<metric>`: - `udfProcessingTime` — a Histogram of per-invocation UDF time (for async UDFs, the full dispatch-to-completion span). - `udfExceptionCount` — a Counter of exceptions escaping the UDF (synchronous exceptions propagating out of `eval`, and asynchronous exceptional completions). The feature is **off by default** (`table.exec.udf-metric-enabled = false`) with **zero overhead when disabled** — the instrumentation is emitted at code generation only when enabled, so the generated operator is byte-identical to today when off. When enabled, it uses the same counter-based sampling as state latency tracking (FLINK-21736): only every Nth invocation is timed (`table.exec.udf-metric.sample-interval`, default 100), while exceptions are counted on every invocation. It is opened as a **draft** to validate the FLIP design against master and give the DISCUSS/VOTE thread a concrete reference; it is not intended to presume the vote outcome. ## Brief change log - Add two `@PublicEvolving` options to `ExecutionConfigOptions`: `table.exec.udf-metric-enabled` (default false) and `table.exec.udf-metric.sample-interval` (default 100), plus generated config docs. - Add a reusable `UdfMetrics` helper (`flink-table-runtime`) that registers the two metrics under `addGroup("udf", udfName)`, owns the FLINK-21736 sampling decision, brackets timing, and counts exceptions. The processing-time histogram is a `DescriptiveStatisticsHistogram` and the exception counter is a `ThreadSafeSimpleCounter`. - Instrument synchronous scalar and table UDF calls at code generation (`BridgingFunctionGenUtil`), sharing one handle per `(operator, udf name)`. Only user functions on the modern `BridgingSqlFunction` stack are metered. - Instrument asynchronous scalar and table UDF calls: the sampling decision and start time are captured at dispatch on the task thread; `udfProcessingTime` and `udfExceptionCount` are recorded at completion on the callback thread, touching only the synchronized histogram and the thread-safe counter. - Document the feature under `docs/…/ops/metrics.md`. Not metered (out of scope of this FLIP): Process Table Functions (`PROCESS_TABLE`), aggregate functions, and functions registered via the deprecated `registerFunction` API. ## Verifying this change This change added tests and can be verified as follows: - `UdfMetricsTest` (unit) — the sampling decision (including the `sample-interval = 1` "every call" case and the reset boundary), timing, and exception counting. - `UdfMetricsITCase` (integration, `InMemoryReporter`) — 12 cases covering sync scalar and table, async scalar and table, metric naming/scope, values (a delayed UDF whose `udfProcessingTime` reflects the delay), exception counting, the enabled/disabled gate, the one-handle-per-`(operator, udf)` sharing, and — for async — an exceptional completion that increments `udfExceptionCount` while the job still finishes. - `ConfigOptionsDocsCompletenessITCase` and the regenerated config docs verify the two new options are documented. - Overhead was sanity-checked with a local micro-benchmark (the disabled path is byte-identical, i.e. zero; the enabled non-sampled fast path is a single integer increment). A rigorous JMH benchmark belongs in `flink-benchmarks`. ## 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)`: **yes** — two new `@PublicEvolving` `ConfigOption`s in `ExecutionConfigOptions`. - The serializers: **no** - The runtime per-record code paths (performance sensitive): **yes** — the instrumentation wraps the generated UDF call site. It is gated at code generation (byte-identical when disabled) and counter-sampled when enabled, so the disabled path has zero overhead and the enabled fast path is a single integer increment. - 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? **docs** (`docs/content/docs/ops/metrics.md` and the Chinese copy) and **JavaDocs**. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (Claude Code, Anthropic Claude Opus 4.8) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
