[FLINK-28804][formats] Use proper stand-ins for missing metric groups

Author: zentol

flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/DeserializationSchemaAdapter.java

@@ -29,6 +29,7 @@
 import org.apache.flink.core.fs.FileInputSplit;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.InstantiationUtil;
@@ -64,8 +65,7 @@ private DeserializationSchema<RowData> createDeserialization() throws IOExceptio
                     new DeserializationSchema.InitializationContext() {
                         @Override
                         public MetricGroup getMetricGroup() {
-                            throw new UnsupportedOperationException(
-                                    "MetricGroup is unsupported in BulkFormat.");
+                            return new UnregisteredMetricsGroup();
                         }
 
                         @Override

flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/SerializationSchemaAdapter.java

@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.serialization.Encoder;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.util.UserCodeClassLoader;
 
@@ -59,8 +60,7 @@ private void checkOpened() throws IOException {
                         new SerializationSchema.InitializationContext() {
                             @Override
                             public MetricGroup getMetricGroup() {
-                                throw new UnsupportedOperationException(
-                                        "MetricGroup is unsupported in BulkFormat.");
+                                return new UnregisteredMetricsGroup();
                             }
 
                             @Override

flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java

@@ -19,10 +19,9 @@
 
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
-import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.connector.testutils.formats.DummyInitializationContext;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.util.TestLogger;
-import org.apache.flink.util.UserCodeClassLoader;
 
 import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
 import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
@@ -351,28 +350,6 @@ public int partition(
     }
 
     private void open(KafkaRecordSerializationSchema<String> schema) throws Exception {
-        schema.open(
-                new SerializationSchema.InitializationContext() {
-                    @Override
-                    public MetricGroup getMetricGroup() {
-                        return null;
-                    }
-
-                    @Override
-                    public UserCodeClassLoader getUserCodeClassLoader() {
-                        return new UserCodeClassLoader() {
-                            @Override
-                            public ClassLoader asClassLoader() {
-                                return KafkaRecordSerializationSchemaBuilderTest.class
-                                        .getClassLoader();
-                            }
-
-                            @Override
-                            public void registerReleaseHookIfAbsent(
-                                    String releaseHookName, Runnable releaseHook) {}
-                        };
-                    }
-                },
-                null);
+        schema.open(new DummyInitializationContext(), null);
     }
 }