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

1. Expose OperatorMetricGroup via RuntimeContext
2. Move the dependency of flink-metric-core from flink-core to flink-core-api

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-process-function-parent/flink-process-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();
 }

flink-process-function-parent/flink-process-function/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();

flink-process-function-parent/flink-process-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;
@@ -38,6 +39,8 @@ public class DefaultRuntimeContext implements RuntimeContext {
 
     private final ProcessingTimeManager processingTimeManager;
 
+    private final OperatorMetricGroup metricGroup;
+
     private final DefaultTimestampManager timestampManager;
 
     public DefaultRuntimeContext(
@@ -52,6 +55,7 @@ public DefaultRuntimeContext(
         this.taskInfo = new DefaultTaskInfo(parallelism, maxParallelism, taskName);
         this.stateManager = new DefaultStateManager(currentKeySupplier, currentKeySetter);
         this.processingTimeManager = processingTimeManager;
+        this.metricGroup = operatorContext.getMetricGroup();
         this.timestampManager = new DefaultTimestampManager();
     }
 
@@ -74,6 +78,11 @@ public ProcessingTimeManager getProcessingTimeManager() {
         return processingTimeManager;
     }
 
+    @Override
+    public OperatorMetricGroup getMetricGroup() {
+        return metricGroup;
+    }
+
     @Override
     public DefaultTimestampManager getTimestampManager() {
         return timestampManager;

flink-process-function-parent/flink-process-function/src/main/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;
@@ -63,6 +64,11 @@ public ProcessingTimeManager getProcessingTimeManager() {
         return UnsupportedProcessingTimeManager.INSTANCE;
     }
 
+    @Override
+    public OperatorMetricGroup getMetricGroup() {
+        return context.getMetricGroup();
+    }
+
     @Override
     public TimestampManager getTimestampManager() {
         return context.getTimestampManager();