(auron) branch master updated: [AURON #2185] Integration with Flink metrics in AuronKafkaSourceFunction (#2186)

[ri...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

richox pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git


The following commit(s) were added to refs/heads/master by this push:
     new f251bbb2 [AURON #2185] Integration with Flink metrics in AuronKafkaSourceFunction (#2186)
f251bbb2 is described below

commit f251bbb21a5396395b13204976d4d68f8e847e59
Author: xuzifu666 <12...@qq.com>
AuthorDate: Fri Apr 10 12:38:24 2026 +0800

    [AURON #2185] Integration with Flink metrics in AuronKafkaSourceFunction (#2186)
    
    # Which issue does this PR close?
    
    Closes https://github.com/apache/auron/issues/2185
    
    # Rationale for this change
    
    # What changes are included in this PR?
    
    # Are there any user-facing changes?
    
    # How was this patch tested?
---
 .../flink/connector/kafka/AuronKafkaSourceFunction.java   | 15 +++++++++++++--
 1 file changed, 13 insertions(+), 2 deletions(-)

diff --git a/auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java b/auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java
index a9bc7c89..178f611e 100644
--- a/auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java
+++ b/auron-flink-extension/auron-flink-runtime/src/main/java/org/apache/auron/flink/connector/kafka/AuronKafkaSourceFunction.java
@@ -50,6 +50,7 @@ import org.apache.flink.api.common.typeinfo.TypeHint;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.runtime.state.FunctionInitializationContext;
 import org.apache.flink.runtime.state.FunctionSnapshotContext;
@@ -117,6 +118,7 @@ public class AuronKafkaSourceFunction extends RichParallelSourceFunction<RowData
     private volatile boolean isRunning;
     private transient String auronOperatorIdWithSubtaskIndex;
     private transient MetricNode nativeMetric;
+    private transient MetricGroup metricGroup;
     private transient ObjectMapper mapper;
 
     // Kafka Consumer for partition metadata discovery only (does NOT consume data)
@@ -282,11 +284,20 @@ public class AuronKafkaSourceFunction extends RichParallelSourceFunction<RowData
 
     @Override
     public void run(SourceContext<RowData> sourceContext) throws Exception {
+        metricGroup = getRuntimeContext().getMetricGroup();
+        final Map<String, Counter> flinkCounters = new HashMap<>();
+
         nativeMetric = new MetricNode(new ArrayList<>()) {
             @Override
             public void add(String name, long value) {
-                // TODO Integration with Flink metrics
-                LOG.info("Metric Auron Source: {} = {}", name, value);
+                // Integration with Flink metrics
+                Counter counter = flinkCounters.get(name);
+                if (counter == null) {
+                    counter = metricGroup.counter(name);
+                    flinkCounters.put(name, counter);
+                }
+                counter.inc(value);
+                LOG.debug("Metric Auron Source: {} = {}", name, value);
             }
         };
         List<RowType.RowField> fieldList = new LinkedList<>();