[API] Expose OperatorMetricGroup via RuntimeContext

Move the dependency of flink-metric-core from flink-core to flink-core-api
reswqa · Feb 28, 2024 · 673c731 · 673c731
1 parent 9d6751a
commit 673c731
Show file tree

Hide file tree

Showing 6 changed files with 31 additions and 6 deletions.
diff --git a/flink-core-api/pom.xml b/flink-core-api/pom.xml
@@ -39,6 +39,12 @@ under the License.
 			<artifactId>flink-annotations</artifactId>
 			<version>${project.version}</version>
 		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-metrics-core</artifactId>
+			<version>${project.version}</version>
+		</dependency>
 	</dependencies>
 
 	<build>

diff --git a/flink-core/pom.xml b/flink-core/pom.xml
@@ -71,12 +71,6 @@ under the License.
 			<version>${project.version}</version>
 		</dependency>
 
-		<dependency>
-			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-metrics-core</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-
 		<!-- ArchUnit test dependencies -->
 
 		<dependency>

diff --git a/...ocess-function-api/src/main/java/org/apache/flink/process/api/context/RuntimeContext.java b/...ocess-function-api/src/main/java/org/apache/flink/process/api/context/RuntimeContext.java
@@ -19,6 +19,7 @@
 package org.apache.flink.process.api.context;
 
 import org.apache.flink.annotation.Experimental;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
 
 /**
  * A RuntimeContext contains information about the context in which process functions are executed.
@@ -41,6 +42,9 @@ public interface RuntimeContext {
     /** Get the {@link ProcessingTimeManager} of this process function. */
     ProcessingTimeManager getProcessingTimeManager();
 
+    /** Get the metric group of this process function. */
+    OperatorMetricGroup getMetricGroup();
+
     /** Get the {@link TimestampManager} of this process function. */
     TimestampManager getTimestampManager();
 }
diff --git a/...ion/src/main/java/org/apache/flink/process/impl/context/DefaultNonPartitionedContext.java b/...ion/src/main/java/org/apache/flink/process/impl/context/DefaultNonPartitionedContext.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.process.impl.context;
 
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
 import org.apache.flink.process.api.context.JobInfo;
 import org.apache.flink.process.api.context.NonPartitionedContext;
 import org.apache.flink.process.api.context.ProcessingTimeManager;
@@ -61,6 +62,11 @@ public ProcessingTimeManager getProcessingTimeManager() {
         return UnsupportedProcessingTimeManager.INSTANCE;
     }
 
+    @Override
+    public OperatorMetricGroup getMetricGroup() {
+        return context.getMetricGroup();
+    }
+
     @Override
     public TimestampManager getTimestampManager() {
         return context.getTimestampManager();

diff --git a/...s-function/src/main/java/org/apache/flink/process/impl/context/DefaultRuntimeContext.java b/...s-function/src/main/java/org/apache/flink/process/impl/context/DefaultRuntimeContext.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.process.impl.context;
 
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
 import org.apache.flink.process.api.context.JobInfo;
 import org.apache.flink.process.api.context.ProcessingTimeManager;
 import org.apache.flink.process.api.context.RuntimeContext;
@@ -37,6 +38,8 @@ public class DefaultRuntimeContext implements RuntimeContext {
 
     private final ProcessingTimeManager processingTimeManager;
 
+    private final OperatorMetricGroup metricGroup;
+
     private final DefaultTimestampManager timestampManager;
 
     public DefaultRuntimeContext(
@@ -50,6 +53,7 @@ public DefaultRuntimeContext(
         this.taskInfo = new DefaultTaskInfo(parallelism, maxParallelism, taskName);
         this.stateManager = new DefaultStateManager(currentKeySupplier);
         this.processingTimeManager = processingTimeManager;
+        this.metricGroup = operatorContext.getMetricGroup();
         this.timestampManager = new DefaultTimestampManager();
     }
 
@@ -72,6 +76,11 @@ public ProcessingTimeManager getProcessingTimeManager() {
         return processingTimeManager;
     }
 
+    @Override
+    public OperatorMetricGroup getMetricGroup() {
+        return metricGroup;
+    }
+
     @Override
     public DefaultTimestampManager getTimestampManager() {
         return timestampManager;

diff --git a/...ain/java/org/apache/flink/process/impl/context/DefaultTwoOutputNonPartitionedContext.java b/...ain/java/org/apache/flink/process/impl/context/DefaultTwoOutputNonPartitionedContext.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.process.impl.context;
 
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
 import org.apache.flink.process.api.context.JobInfo;
 import org.apache.flink.process.api.context.ProcessingTimeManager;
 import org.apache.flink.process.api.context.StateManager;
@@ -61,6 +62,11 @@ public ProcessingTimeManager getProcessingTimeManager() {
         return UnsupportedProcessingTimeManager.INSTANCE;
     }
 
+    @Override
+    public OperatorMetricGroup getMetricGroup() {
+        return context.getMetricGroup();
+    }
+
     @Override
     public TimestampManager getTimestampManager() {
         return context.getTimestampManager();