Skip to content

Navigation Menu

Sign in
Sign up

[AURON #2189] Periodically publish native metrics for long-running Flink tasks - #2494

Open
qingfei1994 wants to merge 4 commits into
apache:master from
qingfei1994:master
Open

[AURON #2189] Periodically publish native metrics for long-running Flink tasks #2494
qingfei1994 wants to merge 4 commits into
apache:master from
qingfei1994:master

Conversation

@qingfei1994

@qingfei1994 qingfei1994 commented Aug 31, 2026 •
edited by cxzl25
Loading

Copy link
Copy Markdown

Which issue does this PR close?

Closes #2189

Rationale for this change

Native DataFusion metrics were only published to MetricNode when the native runtime finalized. Spark batch tasks hit that path; long-running Flink tasks (Kafka source) do not, so Flink counters never moved after #2186 wired add to Flink.

What changes are included in this PR?

  • Periodic native metric publish on a tokio interval, sending positive deltas so incremental MetricNode.add does not double-count.
  • Config auron.metrics.update.interval.ms (default 1000; 0 = finalize-only).
  • FlinkMetricNode.fromPlan so Kafka (including fused Project[Filter[KafkaScan]]) and Calc share a plan-shaped metric tree for getChild(i).

Are there any user-facing changes?

Operators can set auron.metrics.update.interval.ms (default 1000ms). Spark SQL metrics may update during the job as a side effect of the same ticker.

How was this patch tested?

  • cargo test -p auron metrics
  • Maven: FlinkMetricNodeFromPlanTest, AuronKafkaSourceFunctionMergeTest, FlinkMetricNodeTest, AuronAdaptorTest

Was this patch authored or co-authored using generative AI tooling?

  • Yes
  • No

If yes, include: Generated-by: Cursor Grok 4.6

ASF guidance: https://www.apache.org/legal/generative-tooling.html

Copy link
Copy Markdown
Contributor

hi @richox could you please help review this PR when you get a chance? Thanks!

richox commented Sep 1, 2026

Copy link
Copy Markdown
Contributor

Thanks for implementing scheduled metric reporting in Auron.

Since this feature is currently only needed for Flink and not for Spark, I suggest adding a configuration item (e.g., auron.metrics.ticker.enabled) that is disabled by default and only enabled for Flink. This keeps the behavior opt-in and avoids unnecessary overhead for other engines.

Could you update the PR accordingly?
qingfei1994 reacted with thumbs up emoji 

...able metrics update for flink

Copy link
Copy Markdown
Author

@richox I use a new configuration "auron.metrics.update.enabled", which is disabled by default and only when "flink.auron.metrics.update.enabled" = true and "flink.auron.metrics.update.interval.ms">0 will triggering the report metrics feature.

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 Needs a closer look

It introduces cross-thread periodic JNI metric publishing and shutdown coordination in the native runtime, which warrants careful human review of concurrency/lifecycle safety.

Pull request overview

Adds periodic native metric publishing so long-running Flink tasks (e.g., Kafka source) surface DataFusion metrics continuously instead of only on native runtime finalization, while avoiding double-counting by sending positive deltas into incremental MetricNode.add.

Changes:

  • Add a Tokio interval-driven metrics "ticker" in the native runtime and publish plan metrics periodically using a per-runtime snapshot to emit positive deltas only.
  • Introduce new config options auron.metrics.update.enabled and auron.metrics.update.interval.ms (with Flink overriding the enabled default).
  • Unify Flink metric tree construction via FlinkMetricNode.fromPlan so KafkaScan and Calc (including fused shapes) share a plan-mirrored MetricNode.getChild(i) structure.
File summaries
File Description
native-engine/auron/src/rt.rs Spawns/stops a periodic metrics publish task and flushes metrics on finalize.
native-engine/auron/src/metrics.rs Tracks last-published metric values and computes positive deltas per plan-node path.
native-engine/auron-jni-bridge/src/conf.rs Exposes new JNI-reflected conf fields for metrics update enable/interval.
auron-flink-extension/auron-flink-runtime/src/test/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunctionMergeTest.java Adds tests asserting correct metric-tree shape for KafkaScan and fused plans.
auron-flink-extension/auron-flink-runtime/src/test/java/org/apache/auron/flink/configuration/FlinkAuronConfigurationTest.java Verifies Flink-specific default enabling of periodic metric updates.
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/runtime/operator/FlinkAuronCalcOperator.java Switches Calc operator metric tree creation to shared FlinkMetricNode.fromPlan.
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/metric/FlinkMetricNode.java Adds fromPlan builder to mirror supported PhysicalPlanNode shapes into a MetricNode tree.
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java Replaces ad-hoc Flink counter wiring with plan-shaped FlinkMetricNode for native publishing.
auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/configuration/FlinkAuronConfiguration.java Adds Flink override for METRICS_UPDATE_ENABLED so JNI reflection yields Flink-appropriate defaults.
auron-core/src/test/java/org/apache/auron/jni/AuronAdaptorTest.java Extends adaptor tests to cover the new metrics update config defaults.
auron-core/src/main/java/org/apache/auron/configuration/AuronConfiguration.java Defines new public config options for enabling periodic updates and setting the interval.
Review details
  • Files reviewed: 11/11 changed files
  • Comments generated: 1
  • Review effort level: Lite

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Comment on lines +373 to +377
let metrics = jni_call!(
AuronCallNativeWrapper(native_wrapper.as_obj()).getMetrics() -> JObject
)?;
let mut snapshot = metric_state.lock();
update_metric_node(metrics.as_obj(), plan.clone(), &mut snapshot)?;
@cxzl25 cxzl25 changed the title (削除) [AURON #2189] Periodically publish native metrics for long-running Fl... (削除ここまで) (追記) [AURON #2189] Periodically publish native metrics for long-running Flink tasks (追記ここまで) Sep 9, 2026

Copy link
Copy Markdown
Author

@richox @Tartarus0zm could you help review this pr?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

Copilot code review Copilot
Copilot review effort, defaults to Lite
Applies to this pull request for everyone.Learn more about Copilot code review.
Copilot left review comments
@richox richox Awaiting requested review from richox

At least 1 approving review is required to merge this pull request.

Assignees

No one assigned

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

Auron MetricNode periodic updates

AltStyle によって変換されたページ (->オリジナル) /