Flink: DynamicSink: Report writer records/bytes send metrics (#14878)

Author: aiborodin

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriter.java

@@ -145,6 +145,7 @@ public void write(DynamicRecordInternal element, Context context)
               return taskWriterFactory.create();
             })
         .write(element.rowData());
+    metrics.mainMetricsGroup().getNumRecordsSendCounter().inc();
   }
 
   @Override

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriterMetrics.java

@@ -18,24 +18,37 @@
  */
 package org.apache.iceberg.flink.sink.dynamic;
 
+import java.util.Arrays;
 import java.util.Map;
-import org.apache.flink.metrics.MetricGroup;
+import java.util.function.ToLongFunction;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.flink.sink.IcebergStreamWriterMetrics;
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.ScanTaskUtil;
 
 class DynamicWriterMetrics {
 
   private final Map<String, IcebergStreamWriterMetrics> metrics;
-  private final MetricGroup mainMetricsGroup;
+  private final SinkWriterMetricGroup mainMetricsGroup;
 
-  DynamicWriterMetrics(MetricGroup mainMetricsGroup) {
+  DynamicWriterMetrics(SinkWriterMetricGroup mainMetricsGroup) {
     this.mainMetricsGroup = mainMetricsGroup;
     this.metrics = Maps.newHashMap();
   }
 
+  SinkWriterMetricGroup mainMetricsGroup() {
+    return this.mainMetricsGroup;
+  }
+
   public void updateFlushResult(String fullTableName, WriteResult result) {
     writerMetrics(fullTableName).updateFlushResult(result);
+
+    long bytesOutTotal = sum(result.dataFiles()) + sum(result.deleteFiles());
+    this.mainMetricsGroup.getNumBytesSendCounter().inc(bytesOutTotal);
   }
 
   public void flushDuration(String fullTableName, long flushDurationMs) {
@@ -46,4 +59,16 @@ IcebergStreamWriterMetrics writerMetrics(String fullTableName) {
     return metrics.computeIfAbsent(
         fullTableName, tableName -> new IcebergStreamWriterMetrics(mainMetricsGroup, tableName));
   }
+
+  private static long sum(DataFile[] files) {
+    return sum(files, DataFile::fileSizeInBytes);
+  }
+
+  private static long sum(DeleteFile[] files) {
+    return sum(files, ScanTaskUtil::contentSizeInBytes);
+  }
+
+  private static <T extends ContentFile<T>> long sum(T[] files, ToLongFunction<T> sizeExtractor) {
+    return Arrays.stream(files).mapToLong(sizeExtractor).sum();
+  }
 }

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicWriter.java

@@ -243,7 +243,7 @@ void testUniqueFileSuffixOnFactoryRecreation() throws Exception {
             1024L,
             properties,
             100,
-            new DynamicWriterMetrics(new UnregisteredMetricsGroup()),
+            new DynamicWriterMetrics(UnregisteredMetricsGroup.createSinkWriterMetricGroup()),
             0,
             0);
     return dynamicWriter;