Flink: Do not ship optional flink-metrics-dropwizard dependency

docs/docs/flink-writes.md

@@ -207,9 +207,9 @@ They should have the following key-value tags.
 | dataFilesSizeHistogram    | Histogram  | Histogram distribution of data file sizes (in bytes).                                               |
 | deleteFilesSizeHistogram  | Histogram  | Histogram distribution of delete file sizes (in bytes).                                             |
 
-The `Histogram` metrics above require `flink-metrics-dropwizard` on the classpath, which is not shipped
-by Flink by default. When using `iceberg-flink-runtime`, this dependency is already bundled. When using
-the `iceberg-flink` artifact directly, add `org.apache.flink:flink-metrics-dropwizard` as a dependency.
+The `Histogram` metrics above require `org.apache.flink:flink-metrics-dropwizard` on the classpath,
+which is not shipped by Flink by default. Please add this artifact to your classpath to see histogram metrics.
+If not present, histogram metrics will be missing. All other metric types will continue to get published.
 
 Committer metrics are added under the sub group of `IcebergFilesCommitter`.
 They should have the following key-value tags.

flink/v2.1/build.gradle

@@ -169,9 +169,6 @@ project(":iceberg-flink:iceberg-flink-runtime-${flinkMajorVersion}") {
       exclude group: 'com.google.code.findbugs', module: 'jsr305'
     }
 
-    // To support dropwizard histogram metrics (not shipped by Flink by default)
-    implementation libs.flink21.metrics.dropwizard
-
     // for integration testing with the flink-runtime-jar
     // all of those dependencies are required because the integration test extends FlinkTestBase
     integrationCompileOnly project(':iceberg-api')

flink/v2.1/flink-runtime/LICENSE

@@ -460,22 +460,6 @@ License: Apache License, Version 2.0 - https://www.apache.org/licenses/LICENSE-2
 
 --------------------------------------------------------------------------------
 
-This product bundles Dropwizard Metrics.
-
-Copyright: (c) 2010-2013 Coda Hale, Yammer.com, 2014-2021 Dropwizard Team
-Project URL: https://github.com/dropwizard/metrics
-License: Apache License, Version 2.0 - https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product bundles Apache Flink's optional support for Dropwizard Metrics.
-
-Copyright: 2014-2026 The Apache Software Foundation
-Project URL: https://flink.apache.org/
-License: Apache License, Version 2.0 - https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
 This product bundles RoaringBitmap.
 
 Copyright: (c) 2013-... the RoaringBitmap authors

flink/v2.1/flink-runtime/runtime-deps.txt

@@ -6,11 +6,9 @@ com.github.luben:zstd-jni:1.5.7-3
 com.google.errorprone:error_prone_annotations:2.10.0
 dev.failsafe:failsafe:3.3.2
 io.airlift:aircompressor:2.0.3
-io.dropwizard.metrics:metrics-core:3.2.6
 org.apache.avro:avro:1.12.1
 org.apache.datasketches:datasketches-java:6.2.0
 org.apache.datasketches:datasketches-memory:3.0.2
-org.apache.flink:flink-metrics-dropwizard:2.1.0
 org.apache.httpcomponents.client5:httpclient5:5.6
 org.apache.httpcomponents.core5:httpcore5-h2:5.4
 org.apache.httpcomponents.core5:httpcore5:5.4

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

@@ -18,23 +18,33 @@
  */
 package org.apache.iceberg.flink.sink;
 
-import com.codahale.metrics.SlidingWindowReservoir;
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper;
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Histogram;
 import org.apache.flink.metrics.MetricGroup;
+import org.apache.iceberg.common.DynClasses;
+import org.apache.iceberg.common.DynConstructors;
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.util.ScanTaskUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Internal
 public class IcebergStreamWriterMetrics {
+
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergStreamWriterMetrics.class);
+
   // 1,024 reservoir size should cost about 8KB, which is quite small.
   // It should also produce good accuracy for histogram distribution (like percentiles).
   private static final int HISTOGRAM_RESERVOIR_SIZE = 1024;
 
+  // Histogram metrics loaded through Flink's optional flink-metrics-dropwizard dependency.
+  // Will be null if not available.
+  private static final DropwizardCtors DROPWIZARD = loadDropwizardCtors();
+
   private final Counter flushedDataFiles;
   private final Counter flushedDeleteFiles;
   private final Counter flushedReferencedDataFiles;
@@ -51,18 +61,8 @@ public IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName) {
     this.lastFlushDurationMs = new AtomicLong();
     writerMetrics.gauge("lastFlushDurationMs", lastFlushDurationMs::get);
 
-    com.codahale.metrics.Histogram dropwizardDataFilesSizeHistogram =
-        new com.codahale.metrics.Histogram(new SlidingWindowReservoir(HISTOGRAM_RESERVOIR_SIZE));
-    this.dataFilesSizeHistogram =
-        writerMetrics.histogram(
-            "dataFilesSizeHistogram",
-            new DropwizardHistogramWrapper(dropwizardDataFilesSizeHistogram));
-    com.codahale.metrics.Histogram dropwizardDeleteFilesSizeHistogram =
-        new com.codahale.metrics.Histogram(new SlidingWindowReservoir(HISTOGRAM_RESERVOIR_SIZE));
-    this.deleteFilesSizeHistogram =
-        writerMetrics.histogram(
-            "deleteFilesSizeHistogram",
-            new DropwizardHistogramWrapper(dropwizardDeleteFilesSizeHistogram));
+    this.dataFilesSizeHistogram = registerHistogram(writerMetrics, "dataFilesSizeHistogram");
+    this.deleteFilesSizeHistogram = registerHistogram(writerMetrics, "deleteFilesSizeHistogram");
   }
 
   public void updateFlushResult(WriteResult result) {
@@ -74,16 +74,21 @@ public void updateFlushResult(WriteResult result) {
     // This should works equally well and we avoided the overhead of tracking the list of file sizes
     // in the {@link CommitSummary}, which currently stores simple stats for counters and gauges
     // metrics.
-    Arrays.stream(result.dataFiles())
-        .forEach(
-            dataFile -> {
-              dataFilesSizeHistogram.update(dataFile.fileSizeInBytes());
-            });
-    Arrays.stream(result.deleteFiles())
-        .forEach(
-            deleteFile -> {
-              deleteFilesSizeHistogram.update(ScanTaskUtil.contentSizeInBytes(deleteFile));
-            });
+    if (dataFilesSizeHistogram != null) {
+      Arrays.stream(result.dataFiles())
+          .forEach(
+              dataFile -> {
+                dataFilesSizeHistogram.update(dataFile.fileSizeInBytes());
+              });
+    }
+
+    if (deleteFilesSizeHistogram != null) {
+      Arrays.stream(result.deleteFiles())
+          .forEach(
+              deleteFile -> {
+                deleteFilesSizeHistogram.update(ScanTaskUtil.contentSizeInBytes(deleteFile));
+              });
+    }
   }
 
   public void flushDuration(long flushDurationMs) {
@@ -97,4 +102,60 @@ public Counter getFlushedDataFiles() {
   public Counter getFlushedDeleteFiles() {
     return flushedDeleteFiles;
   }
+
+  @VisibleForTesting
+  Histogram dataFilesSizeHistogram() {
+    return dataFilesSizeHistogram;
+  }
+
+  @VisibleForTesting
+  Histogram deleteFilesSizeHistogram() {
+    return deleteFilesSizeHistogram;
+  }
+
+  private static Histogram registerHistogram(MetricGroup group, String name) {
+    Histogram histogram = newDropwizardHistogram();
+    return histogram != null ? group.histogram(name, histogram) : null;
+  }
+
+  private static Histogram newDropwizardHistogram() {
+    if (DROPWIZARD == null) {
+      return null;
+    }
+
+    Object reservoir = DROPWIZARD.reservoirCtor.newInstance(HISTOGRAM_RESERVOIR_SIZE);
+    Object codahaleHistogram = DROPWIZARD.histogramCtor.newInstance(reservoir);
+    return DROPWIZARD.wrapperCtor.newInstance(codahaleHistogram);
+  }
+
+  private static DropwizardCtors loadDropwizardCtors() {
+    try {
+      Class<?> reservoirInterface =
+          DynClasses.builder().impl("com.codahale.metrics.Reservoir").buildChecked();
+      Class<?> codahaleHistogramClass =
+          DynClasses.builder().impl("com.codahale.metrics.Histogram").buildChecked();
+      return new DropwizardCtors(
+          DynConstructors.builder()
+              .impl("com.codahale.metrics.SlidingWindowReservoir", int.class)
+              .buildChecked(),
+          DynConstructors.builder()
+              .impl("com.codahale.metrics.Histogram", reservoirInterface)
+              .buildChecked(),
+          DynConstructors.builder(Histogram.class)
+              .impl(
+                  "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper",
+                  codahaleHistogramClass)
+              .buildChecked());
+    } catch (ClassNotFoundException | NoSuchMethodException e) {
+      LOG.warn(
+          "Cannot load Dropwizard metrics; is org.apache.flink:flink-metrics-dropwizard on the classpath?",
+          e);
+      return null;
+    }
+  }
+
+  private record DropwizardCtors(
+      DynConstructors.Ctor<?> reservoirCtor,
+      DynConstructors.Ctor<?> histogramCtor,
+      DynConstructors.Ctor<Histogram> wrapperCtor) {}
 }

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

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
+
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.iceberg.io.WriteResult;
+import org.junit.jupiter.api.Test;
+
+public class TestIcebergStreamWriterMetrics {
+
+  @Test
+  void histogramsCreatedWhenDropwizardAvailable() {
+    IcebergStreamWriterMetrics metrics =
+        new IcebergStreamWriterMetrics(
+            UnregisteredMetricsGroup.createSinkWriterMetricGroup(), "db.table");
+
+    assertThat(metrics.dataFilesSizeHistogram()).isNotNull();
+    assertThat(metrics.deleteFilesSizeHistogram()).isNotNull();
+
+    assertThatNoException()
+        .isThrownBy(() -> metrics.updateFlushResult(WriteResult.builder().build()));
+  }
+}