weiqingy opened a new pull request, #29007: URL: https://github.com/apache/flink/pull/29007
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](https://issues.apache.org/jira/browse/FLINK-38071). Landing order: | Step | Sub-task | Scope | |---|---|---| | PR-1 ✅ merged | [FLINK-40292](https://issues.apache.org/jira/browse/FLINK-40292) | `UdfMetrics` helper: registration, sampling, timing, exception counting ([#28878](https://github.com/apache/flink/pull/28878)) | | PR-2 🔵 in review | [FLINK-40293](https://issues.apache.org/jira/browse/FLINK-40293) | Config options + sync scalar/table instrumentation ([#28879](https://github.com/apache/flink/pull/28879)) | | **PR-3 (this PR)** | [FLINK-40294](https://issues.apache.org/jira/browse/FLINK-40294) | Async scalar/table instrumentation | | PR-4 | [FLINK-40295](https://issues.apache.org/jira/browse/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? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Anthropic Claude Opus 4.8 and Claude Opus 5) -- 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]
