Flink: missed IcebergSourceReader group in PR #5393 for FLIP-27 sourc…

…e reader metrics (#5401)

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

@@ -137,7 +137,7 @@ public Boundedness getBoundedness() {
 
   @Override
   public SourceReader<T, IcebergSourceSplit> createReader(SourceReaderContext readerContext) {
-    return new IcebergSourceReader<>(readerFunction, readerContext);
+    return new IcebergSourceReader<>(lazyTable().name(), readerFunction, readerContext);
   }
 
   @Override

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceReader.java

@@ -21,19 +21,22 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.connector.source.SourceReaderContext;
 import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
 import org.apache.iceberg.flink.source.split.IcebergSourceSplit;
 import org.apache.iceberg.flink.source.split.SplitRequestEvent;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
+@Internal
 public class IcebergSourceReader<T>
     extends SingleThreadMultiplexSourceReaderBase<
         RecordAndPosition<T>, T, IcebergSourceSplit, IcebergSourceSplit> {
 
-  public IcebergSourceReader(ReaderFunction<T> readerFunction, SourceReaderContext context) {
+  public IcebergSourceReader(
+      String fullTableName, ReaderFunction<T> readerFunction, SourceReaderContext context) {
     super(
-        () -> new IcebergSourceSplitReader<>(readerFunction, context),
+        () -> new IcebergSourceSplitReader<>(fullTableName, readerFunction, context),
         new IcebergSourceRecordEmitter<>(),
         context.getConfiguration(),
         context);

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java

@@ -54,12 +54,13 @@ class IcebergSourceSplitReader<T> implements SplitReader<RecordAndPosition<T>, I
   private IcebergSourceSplit currentSplit;
   private String currentSplitId;
 
-  IcebergSourceSplitReader(ReaderFunction<T> openSplitFunction, SourceReaderContext context) {
+  IcebergSourceSplitReader(
+      String fullTableName, ReaderFunction<T> openSplitFunction, SourceReaderContext context) {
     this.openSplitFunction = openSplitFunction;
     this.indexOfSubtask = context.getIndexOfSubtask();
     this.splits = new ArrayDeque<>();
 
-    MetricGroup metrics = context.metricGroup();
+    MetricGroup metrics = context.metricGroup().addGroup("IcebergSourceReader", fullTableName);
     this.assignedSplits = metrics.counter("assignedSplits");
     this.assignedBytes = metrics.counter("assignedBytes");
     this.finishedSplits = metrics.counter("finishedSplits");

flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java

@@ -137,7 +137,7 @@ public Boundedness getBoundedness() {
 
   @Override
   public SourceReader<T, IcebergSourceSplit> createReader(SourceReaderContext readerContext) {
-    return new IcebergSourceReader<>(readerFunction, readerContext);
+    return new IcebergSourceReader<>(lazyTable().name(), readerFunction, readerContext);
   }
 
   @Override

flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceReader.java

@@ -21,19 +21,22 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.connector.source.SourceReaderContext;
 import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
 import org.apache.iceberg.flink.source.split.IcebergSourceSplit;
 import org.apache.iceberg.flink.source.split.SplitRequestEvent;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
+@Internal
 public class IcebergSourceReader<T>
     extends SingleThreadMultiplexSourceReaderBase<
         RecordAndPosition<T>, T, IcebergSourceSplit, IcebergSourceSplit> {
 
-  public IcebergSourceReader(ReaderFunction<T> readerFunction, SourceReaderContext context) {
+  public IcebergSourceReader(
+      String fullTableName, ReaderFunction<T> readerFunction, SourceReaderContext context) {
     super(
-        () -> new IcebergSourceSplitReader<>(readerFunction, context),
+        () -> new IcebergSourceSplitReader<>(fullTableName, readerFunction, context),
         new IcebergSourceRecordEmitter<>(),
         context.getConfiguration(),
         context);

flink/v1.15/flink/src/main/java/org/apache/iceberg/flink/source/reader/IcebergSourceSplitReader.java

@@ -54,12 +54,13 @@ class IcebergSourceSplitReader<T> implements SplitReader<RecordAndPosition<T>, I
   private IcebergSourceSplit currentSplit;
   private String currentSplitId;
 
-  IcebergSourceSplitReader(ReaderFunction<T> openSplitFunction, SourceReaderContext context) {
+  IcebergSourceSplitReader(
+      String fullTableName, ReaderFunction<T> openSplitFunction, SourceReaderContext context) {
     this.openSplitFunction = openSplitFunction;
     this.indexOfSubtask = context.getIndexOfSubtask();
     this.splits = new ArrayDeque<>();
 
-    MetricGroup metrics = context.metricGroup();
+    MetricGroup metrics = context.metricGroup().addGroup("IcebergSourceReader", fullTableName);
     this.assignedSplits = metrics.counter("assignedSplits");
     this.assignedBytes = metrics.counter("assignedBytes");
     this.finishedSplits = metrics.counter("finishedSplits");