[FLINK-34549][API] Expose MetricGroup via RuntimeContext

Author: reswqa

flink-core-api/pom.xml

@@ -40,6 +40,13 @@ under the License.
 			<version>${project.version}</version>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-metrics-core</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+
 		<!-- ================== test dependencies ================== -->
 
 		<dependency>

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>

flink-datastream-api/src/main/java/org/apache/flink/datastream/api/context/RuntimeContext.java

@@ -19,6 +19,7 @@
 package org.apache.flink.datastream.api.context;
 
 import org.apache.flink.annotation.Experimental;
+import org.apache.flink.metrics.MetricGroup;
 
 /**
  * A RuntimeContext contains information about the context in which process functions are executed.
@@ -33,4 +34,7 @@ public interface RuntimeContext {
 
     /** Get the {@link TaskInfo} of this process function. */
     TaskInfo getTaskInfo();
+
+    /** Get the metric group of this process function. */
+    MetricGroup getMetricGroup();
 }

flink-datastream/src/main/java/org/apache/flink/datastream/impl/context/DefaultNonPartitionedContext.java

@@ -22,6 +22,7 @@
 import org.apache.flink.datastream.api.context.NonPartitionedContext;
 import org.apache.flink.datastream.api.context.TaskInfo;
 import org.apache.flink.datastream.api.function.ApplyPartitionFunction;
+import org.apache.flink.metrics.MetricGroup;
 
 /** The default implementation of {@link NonPartitionedContext}. */
 public class DefaultNonPartitionedContext<OUT> implements NonPartitionedContext<OUT> {
@@ -45,4 +46,9 @@ public JobInfo getJobInfo() {
     public TaskInfo getTaskInfo() {
         return context.getTaskInfo();
     }
+
+    @Override
+    public MetricGroup getMetricGroup() {
+        return context.getMetricGroup();
+    }
 }

flink-datastream/src/main/java/org/apache/flink/datastream/impl/context/DefaultPartitionedContext.java

@@ -23,6 +23,7 @@
 import org.apache.flink.datastream.api.context.ProcessingTimeManager;
 import org.apache.flink.datastream.api.context.RuntimeContext;
 import org.apache.flink.datastream.api.context.TaskInfo;
+import org.apache.flink.metrics.MetricGroup;
 
 import java.util.function.Consumer;
 import java.util.function.Supplier;
@@ -64,4 +65,9 @@ public DefaultStateManager getStateManager() {
     public ProcessingTimeManager getProcessingTimeManager() {
         return processingTimeManager;
     }
+
+    @Override
+    public MetricGroup getMetricGroup() {
+        return context.getMetricGroup();
+    }
 }

flink-datastream/src/main/java/org/apache/flink/datastream/impl/context/DefaultRuntimeContext.java

@@ -21,6 +21,7 @@
 import org.apache.flink.datastream.api.context.JobInfo;
 import org.apache.flink.datastream.api.context.RuntimeContext;
 import org.apache.flink.datastream.api.context.TaskInfo;
+import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.runtime.jobgraph.JobType;
 
 /** The default implementation of {@link RuntimeContext}. */
@@ -29,10 +30,18 @@ public class DefaultRuntimeContext implements RuntimeContext {
 
     private final DefaultTaskInfo taskInfo;
 
+    private final MetricGroup metricGroup;
+
     public DefaultRuntimeContext(
-            String jobName, JobType jobType, int parallelism, int maxParallelism, String taskName) {
+            String jobName,
+            JobType jobType,
+            int parallelism,
+            int maxParallelism,
+            String taskName,
+            MetricGroup metricGroup) {
         this.jobInfo = new DefaultJobInfo(jobName, jobType);
         this.taskInfo = new DefaultTaskInfo(parallelism, maxParallelism, taskName);
+        this.metricGroup = metricGroup;
     }
 
     @Override
@@ -44,4 +53,9 @@ public JobInfo getJobInfo() {
     public TaskInfo getTaskInfo() {
         return taskInfo;
     }
+
+    @Override
+    public MetricGroup getMetricGroup() {
+        return metricGroup;
+    }
 }

flink-datastream/src/main/java/org/apache/flink/datastream/impl/context/DefaultTwoOutputNonPartitionedContext.java

@@ -22,6 +22,7 @@
 import org.apache.flink.datastream.api.context.TaskInfo;
 import org.apache.flink.datastream.api.context.TwoOutputNonPartitionedContext;
 import org.apache.flink.datastream.api.function.TwoOutputApplyPartitionFunction;
+import org.apache.flink.metrics.MetricGroup;
 
 /** The default implementation of {@link TwoOutputNonPartitionedContext}. */
 public class DefaultTwoOutputNonPartitionedContext<OUT1, OUT2>
@@ -47,4 +48,9 @@ public JobInfo getJobInfo() {
     public TaskInfo getTaskInfo() {
         return context.getTaskInfo();
     }
+
+    @Override
+    public MetricGroup getMetricGroup() {
+        return context.getMetricGroup();
+    }
 }

flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/ProcessOperator.java

@@ -64,7 +64,8 @@ public void open() throws Exception {
                         operatorContext.getJobType(),
                         taskInfo.getNumberOfParallelSubtasks(),
                         taskInfo.getMaxNumberOfParallelSubtasks(),
-                        taskInfo.getTaskName());
+                        taskInfo.getTaskName(),
+                        operatorContext.getMetricGroup());
         partitionedContext =
                 new DefaultPartitionedContext(
                         context, this::currentKey, this::setCurrentKey, getProcessingTimeManager());

flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputBroadcastProcessOperator.java

@@ -68,7 +68,8 @@ public void open() throws Exception {
                         operatorContext.getJobType(),
                         taskInfo.getNumberOfParallelSubtasks(),
                         taskInfo.getMaxNumberOfParallelSubtasks(),
-                        taskInfo.getTaskName());
+                        taskInfo.getTaskName(),
+                        operatorContext.getMetricGroup());
         this.partitionedContext =
                 new DefaultPartitionedContext(
                         context, this::currentKey, this::setCurrentKey, getProcessingTimeManager());

flink-datastream/src/main/java/org/apache/flink/datastream/impl/operators/TwoInputNonBroadcastProcessOperator.java

@@ -68,7 +68,8 @@ public void open() throws Exception {
                         operatorContext.getJobType(),
                         taskInfo.getNumberOfParallelSubtasks(),
                         taskInfo.getMaxNumberOfParallelSubtasks(),
-                        taskInfo.getTaskName());
+                        taskInfo.getTaskName(),
+                        operatorContext.getMetricGroup());
         this.partitionedContext =
                 new DefaultPartitionedContext(
                         context, this::currentKey, this::setCurrentKey, getProcessingTimeManager());