From e3b754703b813947cc4838e627df8e4f1c697d91 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Wed, 29 Apr 2026 11:40:01 +0200 Subject: [PATCH 01/17] Flink: Do not ship optional flink-metrics-dropwizard dependency Flink does not bundle flink-metrics-dropwizard, which enables histogram metrics. Users need to manually include the dependency for histograms to work. In Iceberg, we used to include the dependency in the flink-runtime jar. Ryan had removed the dependency in #16093 to avoid LICENSE issues, but we added it back in #16126 with the corresponding LICENSE changes, to avoid breaking changes for users. After discussing with Peter, we decided to remove the bundling going forward. This PR makes sure there are no regressions removing the dependency. I've updated the docs and the changelog to inform users of the change. References: - https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/metrics/#histogram - https://github.com/apache/flink/blob/ccf45727eeb2e920f7939f453db67a411e3cb109/flink-dist/pom.xml#L284 --- docs/docs/flink-writes.md | 6 +- flink/v2.1/build.gradle | 3 - flink/v2.1/flink-runtime/LICENSE | 16 --- flink/v2.1/flink-runtime/runtime-deps.txt | 2 - .../sink/IcebergStreamWriterMetrics.java | 103 ++++++++++++++---- .../sink/TestIcebergStreamWriterMetrics.java | 72 ++++++++++++ site/docs/releases.md | 5 + 7 files changed, 162 insertions(+), 45 deletions(-) create mode 100644 flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java diff --git a/docs/docs/flink-writes.md b/docs/docs/flink-writes.md index c904635d0dbb..03795b5beed0 100644 --- a/docs/docs/flink-writes.md +++ b/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. diff --git a/flink/v2.1/build.gradle b/flink/v2.1/build.gradle index 6dc373e6b566..11e7bccb2557 100644 --- a/flink/v2.1/build.gradle +++ b/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') diff --git a/flink/v2.1/flink-runtime/LICENSE b/flink/v2.1/flink-runtime/LICENSE index 11460c3307c8..364652a5aca2 100644 --- a/flink/v2.1/flink-runtime/LICENSE +++ b/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 diff --git a/flink/v2.1/flink-runtime/runtime-deps.txt b/flink/v2.1/flink-runtime/runtime-deps.txt index 3dfc56f15ea9..00c53ed388d0 100644 --- a/flink/v2.1/flink-runtime/runtime-deps.txt +++ b/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 diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index 434f3969577f..a5092b44769a 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -18,19 +18,25 @@ */ package org.apache.iceberg.flink.sink; -import com.codahale.metrics.SlidingWindowReservoir; import java.util.Arrays; import java.util.concurrent.atomic.AtomicLong; +import javax.annotation.Nullable; import org.apache.flink.annotation.Internal; +import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper; import org.apache.flink.metrics.Counter; import org.apache.flink.metrics.Histogram; import org.apache.flink.metrics.MetricGroup; 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; @@ -43,6 +49,11 @@ public class IcebergStreamWriterMetrics { private final Histogram deleteFilesSizeHistogram; public IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName) { + this(metrics, fullTableName, IcebergStreamWriterMetrics.class.getClassLoader()); + } + + @VisibleForTesting + IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName, ClassLoader classLoader) { MetricGroup writerMetrics = metrics.addGroup("IcebergStreamWriter").addGroup("table", fullTableName); this.flushedDataFiles = writerMetrics.counter("flushedDataFiles"); @@ -51,18 +62,12 @@ 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)); + loadHistogramIfAvailable( + writerMetrics, "dataFilesSizeHistogram", HISTOGRAM_RESERVOIR_SIZE, classLoader); this.deleteFilesSizeHistogram = - writerMetrics.histogram( - "deleteFilesSizeHistogram", - new DropwizardHistogramWrapper(dropwizardDeleteFilesSizeHistogram)); + loadHistogramIfAvailable( + writerMetrics, "deleteFilesSizeHistogram", HISTOGRAM_RESERVOIR_SIZE, classLoader); } public void updateFlushResult(WriteResult result) { @@ -74,16 +79,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 +107,55 @@ public Counter getFlushedDataFiles() { public Counter getFlushedDeleteFiles() { return flushedDeleteFiles; } + + @Nullable + Histogram getDataFilesSizeHistogram() { + return dataFilesSizeHistogram; + } + + @Nullable + Histogram getDeleteFilesSizeHistogram() { + return deleteFilesSizeHistogram; + } + + static Histogram loadHistogramIfAvailable( + MetricGroup group, String name, int reservoirSize, ClassLoader classLoader) { + + if (isFlinkDropwizardAvailable(classLoader)) { + return HistogramLoader.load(name, group, reservoirSize); + } else { + LOG.warn( + "flink-metrics-dropwizard is not on the classpath. '{}' histogram metrics will be disabled. Add org.apache.flink:flink-metrics-dropwizard to enable them.", + name); + return null; + } + } + + /** + * Checks whether the Dropwizard-based histogram wrapper provided through Flink's optional + * flink-metrics-dropwizard dependency is available. + */ + private static boolean isFlinkDropwizardAvailable(ClassLoader classLoader) { + try { + Class.forName( + "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", false, classLoader); + return true; + } catch (ClassNotFoundException e) { + return false; + } + } + + /** + * Must be encapsulated into a separate class to avoid the JVM eagerly loading any non-existing + * referenced classes. This has been manually verified. + */ + private static final class HistogramLoader { + + static Histogram load(String name, MetricGroup group, int reservoirSize) { + com.codahale.metrics.Histogram codahaleHistogram = + new com.codahale.metrics.Histogram( + new com.codahale.metrics.SlidingWindowReservoir(reservoirSize)); + return group.histogram(name, new DropwizardHistogramWrapper(codahaleHistogram)); + } + } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java new file mode 100644 index 000000000000..53780e2083f6 --- /dev/null +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java @@ -0,0 +1,72 @@ +/* + * 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 java.net.URL; +import java.net.URLClassLoader; +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.getDataFilesSizeHistogram()).isNotNull(); + assertThat(metrics.getDeleteFilesSizeHistogram()).isNotNull(); + + assertThatNoException() + .isThrownBy(() -> metrics.updateFlushResult(WriteResult.builder().build())); + } + + @Test + void histogramsSkippedWhenDropwizardMissing() throws Exception { + try (EmptyClassloader loader = new EmptyClassloader()) { + + IcebergStreamWriterMetrics metrics = + new IcebergStreamWriterMetrics( + UnregisteredMetricsGroup.createSinkWriterMetricGroup(), "db.table", loader); + + assertThat(metrics.getDataFilesSizeHistogram()).isNull(); + assertThat(metrics.getDeleteFilesSizeHistogram()).isNull(); + + assertThatNoException() + .isThrownBy(() -> metrics.updateFlushResult(WriteResult.builder().build())); + } + } + + private static final class EmptyClassloader extends URLClassLoader { + + EmptyClassloader() { + super(new URL[0]); + } + + @Override + protected Class loadClass(String name, boolean resolve) throws ClassNotFoundException { + throw new ClassNotFoundException(name); + } + } +} diff --git a/site/docs/releases.md b/site/docs/releases.md index bc8957942fc2..d27eb4f09135 100644 --- a/site/docs/releases.md +++ b/site/docs/releases.md @@ -67,6 +67,11 @@ To add a dependency on Iceberg in Maven, add the following to your `pom.xml`: ``` +### 1.11.0 release + +* Flink + - Flink's optional `flink-metrics-dropwizard` dependency is no longer shipped in the `iceberg-flink-runtime` fat jar. Without this dependency, the histogram metrics "dataFilesSizeHistogram" and "deleteFilesSizeHistogram" will no longer be published. Users can add manually add back the dependency for the metrics to continue to get published. + ### 1.10.1 release Apache Iceberg 1.10.1 was released on Dec 22, 2025. From 4abf18eb6d64edf56dc51bbc27cc207a57d2f343 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Wed, 29 Apr 2026 15:36:23 +0200 Subject: [PATCH 02/17] fixup! remove get prefix --- .../iceberg/flink/sink/IcebergStreamWriterMetrics.java | 4 ++-- .../flink/sink/TestIcebergStreamWriterMetrics.java | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index a5092b44769a..9e59876f0474 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -109,12 +109,12 @@ public Counter getFlushedDeleteFiles() { } @Nullable - Histogram getDataFilesSizeHistogram() { + Histogram dataFilesSizeHistogram() { return dataFilesSizeHistogram; } @Nullable - Histogram getDeleteFilesSizeHistogram() { + Histogram deleteFilesSizeHistogram() { return deleteFilesSizeHistogram; } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java index 53780e2083f6..b50f4269285f 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java @@ -35,8 +35,8 @@ void histogramsCreatedWhenDropwizardAvailable() { new IcebergStreamWriterMetrics( UnregisteredMetricsGroup.createSinkWriterMetricGroup(), "db.table"); - assertThat(metrics.getDataFilesSizeHistogram()).isNotNull(); - assertThat(metrics.getDeleteFilesSizeHistogram()).isNotNull(); + assertThat(metrics.dataFilesSizeHistogram()).isNotNull(); + assertThat(metrics.deleteFilesSizeHistogram()).isNotNull(); assertThatNoException() .isThrownBy(() -> metrics.updateFlushResult(WriteResult.builder().build())); @@ -50,8 +50,8 @@ void histogramsSkippedWhenDropwizardMissing() throws Exception { new IcebergStreamWriterMetrics( UnregisteredMetricsGroup.createSinkWriterMetricGroup(), "db.table", loader); - assertThat(metrics.getDataFilesSizeHistogram()).isNull(); - assertThat(metrics.getDeleteFilesSizeHistogram()).isNull(); + assertThat(metrics.dataFilesSizeHistogram()).isNull(); + assertThat(metrics.deleteFilesSizeHistogram()).isNull(); assertThatNoException() .isThrownBy(() -> metrics.updateFlushResult(WriteResult.builder().build())); From f519d2cce0a3013e312a8de0f9d7167b761831d3 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Thu, 30 Apr 2026 08:47:08 +0200 Subject: [PATCH 03/17] fixup! inline method --- .../sink/IcebergStreamWriterMetrics.java | 25 +++++++------------ 1 file changed, 9 insertions(+), 16 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index 9e59876f0474..ebb243cfd11b 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -118,30 +118,23 @@ Histogram deleteFilesSizeHistogram() { return deleteFilesSizeHistogram; } - static Histogram loadHistogramIfAvailable( - MetricGroup group, String name, int reservoirSize, ClassLoader classLoader) { - - if (isFlinkDropwizardAvailable(classLoader)) { - return HistogramLoader.load(name, group, reservoirSize); - } else { - LOG.warn( - "flink-metrics-dropwizard is not on the classpath. '{}' histogram metrics will be disabled. Add org.apache.flink:flink-metrics-dropwizard to enable them.", - name); - return null; - } - } - /** * Checks whether the Dropwizard-based histogram wrapper provided through Flink's optional * flink-metrics-dropwizard dependency is available. */ - private static boolean isFlinkDropwizardAvailable(ClassLoader classLoader) { + @SuppressWarnings("CatchBlockLogException") + static Histogram loadHistogramIfAvailable( + MetricGroup group, String name, int reservoirSize, ClassLoader classLoader) { + try { Class.forName( "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", false, classLoader); - return true; + return HistogramLoader.load(name, group, reservoirSize); } catch (ClassNotFoundException e) { - return false; + LOG.warn( + "flink-metrics-dropwizard is not on the classpath. '{}' histogram metrics will be disabled. Add org.apache.flink:flink-metrics-dropwizard to enable them.", + name); + return null; } } From e638ad801524868a9c8b3ff5030f587b3e7a7a42 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Thu, 30 Apr 2026 08:50:12 +0200 Subject: [PATCH 04/17] fixup! tighten method visibility --- .../apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index ebb243cfd11b..32b58667ec21 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -123,7 +123,7 @@ Histogram deleteFilesSizeHistogram() { * flink-metrics-dropwizard dependency is available. */ @SuppressWarnings("CatchBlockLogException") - static Histogram loadHistogramIfAvailable( + private static Histogram loadHistogramIfAvailable( MetricGroup group, String name, int reservoirSize, ClassLoader classLoader) { try { From 8f854aafd2ea47be276facdb63296ee406e49795 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Thu, 30 Apr 2026 09:21:06 +0200 Subject: [PATCH 05/17] fixup! Add @VisibleForTesting --- .../apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index 32b58667ec21..b18835a45a8a 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -108,11 +108,13 @@ public Counter getFlushedDeleteFiles() { return flushedDeleteFiles; } + @VisibleForTesting @Nullable Histogram dataFilesSizeHistogram() { return dataFilesSizeHistogram; } + @VisibleForTesting @Nullable Histogram deleteFilesSizeHistogram() { return deleteFilesSizeHistogram; From 8c00e4aed4bc0887eab82f7e90fa5dd48029ef27 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Thu, 30 Apr 2026 08:54:43 +0200 Subject: [PATCH 06/17] fixup! Use DynClasses to replace Class.forName --- .../flink/sink/IcebergStreamWriterMetrics.java | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index b18835a45a8a..9492fbd50880 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -23,10 +23,10 @@ import javax.annotation.Nullable; import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.VisibleForTesting; -import org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper; 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.io.WriteResult; import org.apache.iceberg.util.ScanTaskUtil; import org.slf4j.Logger; @@ -128,16 +128,20 @@ Histogram deleteFilesSizeHistogram() { private static Histogram loadHistogramIfAvailable( MetricGroup group, String name, int reservoirSize, ClassLoader classLoader) { - try { - Class.forName( - "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", false, classLoader); - return HistogramLoader.load(name, group, reservoirSize); - } catch (ClassNotFoundException e) { + Class wrapperClass = + DynClasses.builder() + .loader(classLoader) + .impl("org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper") + .orNull() + .build(); + if (wrapperClass == null) { LOG.warn( "flink-metrics-dropwizard is not on the classpath. '{}' histogram metrics will be disabled. Add org.apache.flink:flink-metrics-dropwizard to enable them.", name); return null; } + + return HistogramLoader.load(name, group, reservoirSize); } /** From bc07ad2ba1b18a36623ce89e43c50dd3ae0c5b41 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Thu, 30 Apr 2026 08:55:34 +0200 Subject: [PATCH 07/17] fixup! Use Dyn* classes for everything --- .../sink/IcebergStreamWriterMetrics.java | 65 ++++++++++++------- 1 file changed, 42 insertions(+), 23 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index 9492fbd50880..fd9be6a2ffd6 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -27,6 +27,7 @@ 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; @@ -128,33 +129,51 @@ Histogram deleteFilesSizeHistogram() { private static Histogram loadHistogramIfAvailable( MetricGroup group, String name, int reservoirSize, ClassLoader classLoader) { - Class wrapperClass = - DynClasses.builder() - .loader(classLoader) - .impl("org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper") - .orNull() - .build(); - if (wrapperClass == null) { + try { + Class reservoirInterface = + DynClasses.builder() + .loader(classLoader) + .impl("com.codahale.metrics.Reservoir") + .buildChecked(); + Class slidingWindowReservoirClass = + DynClasses.builder() + .loader(classLoader) + .impl("com.codahale.metrics.SlidingWindowReservoir") + .buildChecked(); + Class codahaleHistogramClass = + DynClasses.builder() + .loader(classLoader) + .impl("com.codahale.metrics.Histogram") + .buildChecked(); + Class wrapperClass = + DynClasses.builder() + .loader(classLoader) + .impl("org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper") + .buildChecked(); + + Object reservoir = + DynConstructors.builder() + .impl(slidingWindowReservoirClass, int.class) + .buildChecked() + .newInstance(reservoirSize); + Object codahaleHistogram = + DynConstructors.builder() + .impl(codahaleHistogramClass, reservoirInterface) + .buildChecked() + .newInstance(reservoir); + Histogram wrapper = + (Histogram) + DynConstructors.builder() + .impl(wrapperClass, codahaleHistogramClass) + .buildChecked() + .newInstance(codahaleHistogram); + + return group.histogram(name, wrapper); + } catch (ClassNotFoundException | NoSuchMethodException e) { LOG.warn( "flink-metrics-dropwizard is not on the classpath. '{}' histogram metrics will be disabled. Add org.apache.flink:flink-metrics-dropwizard to enable them.", name); return null; } - - return HistogramLoader.load(name, group, reservoirSize); - } - - /** - * Must be encapsulated into a separate class to avoid the JVM eagerly loading any non-existing - * referenced classes. This has been manually verified. - */ - private static final class HistogramLoader { - - static Histogram load(String name, MetricGroup group, int reservoirSize) { - com.codahale.metrics.Histogram codahaleHistogram = - new com.codahale.metrics.Histogram( - new com.codahale.metrics.SlidingWindowReservoir(reservoirSize)); - return group.histogram(name, new DropwizardHistogramWrapper(codahaleHistogram)); - } } } From e8a911a3c11c094e9102909bc33281e0463f4bfb Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Thu, 30 Apr 2026 09:26:30 +0200 Subject: [PATCH 08/17] fixup! remove changelog entry --- site/docs/releases.md | 5 ----- 1 file changed, 5 deletions(-) diff --git a/site/docs/releases.md b/site/docs/releases.md index d27eb4f09135..bc8957942fc2 100644 --- a/site/docs/releases.md +++ b/site/docs/releases.md @@ -67,11 +67,6 @@ To add a dependency on Iceberg in Maven, add the following to your `pom.xml`: ``` -### 1.11.0 release - -* Flink - - Flink's optional `flink-metrics-dropwizard` dependency is no longer shipped in the `iceberg-flink-runtime` fat jar. Without this dependency, the histogram metrics "dataFilesSizeHistogram" and "deleteFilesSizeHistogram" will no longer be published. Users can add manually add back the dependency for the metrics to continue to get published. - ### 1.10.1 release Apache Iceberg 1.10.1 was released on Dec 22, 2025. From fd6cdea2f1f013ace3253cbce39ae7becb7b3c3f Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Thu, 30 Apr 2026 10:04:42 +0200 Subject: [PATCH 09/17] fixup! re-trigger CI From 2175b2404f62d0160e74fd9ecb95c3077a4ad3fd Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 4 May 2026 05:48:58 +0200 Subject: [PATCH 10/17] fixup! Drop @Nullable on histogram getters --- .../apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index fd9be6a2ffd6..fd0a25546c89 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -20,7 +20,6 @@ import java.util.Arrays; import java.util.concurrent.atomic.AtomicLong; -import javax.annotation.Nullable; import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.metrics.Counter; @@ -110,13 +109,11 @@ public Counter getFlushedDeleteFiles() { } @VisibleForTesting - @Nullable Histogram dataFilesSizeHistogram() { return dataFilesSizeHistogram; } @VisibleForTesting - @Nullable Histogram deleteFilesSizeHistogram() { return deleteFilesSizeHistogram; } From eafdbbe3507a09fb478ca66ac76874f7f8bd783a Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 4 May 2026 05:50:07 +0200 Subject: [PATCH 11/17] fixup! Use DynConstructors.impl string form to drop impl-class lookups --- .../flink/sink/IcebergStreamWriterMetrics.java | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index fd0a25546c89..b0fdb39fa5b0 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -132,25 +132,16 @@ private static Histogram loadHistogramIfAvailable( .loader(classLoader) .impl("com.codahale.metrics.Reservoir") .buildChecked(); - Class slidingWindowReservoirClass = - DynClasses.builder() - .loader(classLoader) - .impl("com.codahale.metrics.SlidingWindowReservoir") - .buildChecked(); Class codahaleHistogramClass = DynClasses.builder() .loader(classLoader) .impl("com.codahale.metrics.Histogram") .buildChecked(); - Class wrapperClass = - DynClasses.builder() - .loader(classLoader) - .impl("org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper") - .buildChecked(); Object reservoir = DynConstructors.builder() - .impl(slidingWindowReservoirClass, int.class) + .loader(classLoader) + .impl("com.codahale.metrics.SlidingWindowReservoir", int.class) .buildChecked() .newInstance(reservoirSize); Object codahaleHistogram = @@ -161,7 +152,10 @@ private static Histogram loadHistogramIfAvailable( Histogram wrapper = (Histogram) DynConstructors.builder() - .impl(wrapperClass, codahaleHistogramClass) + .loader(classLoader) + .impl( + "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", + codahaleHistogramClass) .buildChecked() .newInstance(codahaleHistogram); From b954ec12f580774de9f2c2661ea4e05304d77be0 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 4 May 2026 05:51:11 +0200 Subject: [PATCH 12/17] fixup! Use typed DynConstructors builder for Histogram wrapper --- .../flink/sink/IcebergStreamWriterMetrics.java | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index b0fdb39fa5b0..ce6471983839 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -150,14 +150,13 @@ private static Histogram loadHistogramIfAvailable( .buildChecked() .newInstance(reservoir); Histogram wrapper = - (Histogram) - DynConstructors.builder() - .loader(classLoader) - .impl( - "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", - codahaleHistogramClass) - .buildChecked() - .newInstance(codahaleHistogram); + DynConstructors.builder(Histogram.class) + .loader(classLoader) + .impl( + "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", + codahaleHistogramClass) + .buildChecked() + .newInstance(codahaleHistogram); return group.histogram(name, wrapper); } catch (ClassNotFoundException | NoSuchMethodException e) { From 91491f8a1123700c568f3fc1c48f086cde1f9f38 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 4 May 2026 05:54:51 +0200 Subject: [PATCH 13/17] fixup! Move dropwizard reflection to static fields and warn only once --- .../sink/IcebergStreamWriterMetrics.java | 99 ++++++++++--------- .../sink/TestIcebergStreamWriterMetrics.java | 30 ------ 2 files changed, 51 insertions(+), 78 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index ce6471983839..acd56b5c6116 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -25,7 +25,6 @@ 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; @@ -41,6 +40,48 @@ public class IcebergStreamWriterMetrics { // 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 + private static final boolean DROPWIZARD_AVAILABLE; + private static final DynConstructors.Ctor RESERVOIR_CTOR; + private static final DynConstructors.Ctor CODAHALE_HISTOGRAM_CTOR; + private static final DynConstructors.Ctor WRAPPER_CTOR; + + static { + DynConstructors.Ctor reservoirCtor = null; + DynConstructors.Ctor codahaleHistogramCtor = null; + DynConstructors.Ctor wrapperCtor = null; + boolean available = false; + + try { + Class reservoirInterface = Class.forName("com.codahale.metrics.Reservoir"); + Class codahaleHistogramClass = Class.forName("com.codahale.metrics.Histogram"); + reservoirCtor = + DynConstructors.builder() + .impl("com.codahale.metrics.SlidingWindowReservoir", int.class) + .buildChecked(); + codahaleHistogramCtor = + DynConstructors.builder() + .impl("com.codahale.metrics.Histogram", reservoirInterface) + .buildChecked(); + wrapperCtor = + DynConstructors.builder(Histogram.class) + .impl( + "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", + codahaleHistogramClass) + .buildChecked(); + available = true; + } catch (ClassNotFoundException | NoSuchMethodException e) { + LOG.warn( + "flink-metrics-dropwizard is not on the classpath. Histogram metrics will be disabled. Add org.apache.flink:flink-metrics-dropwizard to enable them.", + e); + } + + RESERVOIR_CTOR = reservoirCtor; + CODAHALE_HISTOGRAM_CTOR = codahaleHistogramCtor; + WRAPPER_CTOR = wrapperCtor; + DROPWIZARD_AVAILABLE = available; + } + private final Counter flushedDataFiles; private final Counter flushedDeleteFiles; private final Counter flushedReferencedDataFiles; @@ -49,11 +90,6 @@ public class IcebergStreamWriterMetrics { private final Histogram deleteFilesSizeHistogram; public IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName) { - this(metrics, fullTableName, IcebergStreamWriterMetrics.class.getClassLoader()); - } - - @VisibleForTesting - IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName, ClassLoader classLoader) { MetricGroup writerMetrics = metrics.addGroup("IcebergStreamWriter").addGroup("table", fullTableName); this.flushedDataFiles = writerMetrics.counter("flushedDataFiles"); @@ -63,11 +99,10 @@ public IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName) { writerMetrics.gauge("lastFlushDurationMs", lastFlushDurationMs::get); this.dataFilesSizeHistogram = - loadHistogramIfAvailable( - writerMetrics, "dataFilesSizeHistogram", HISTOGRAM_RESERVOIR_SIZE, classLoader); + loadHistogramIfAvailable(writerMetrics, "dataFilesSizeHistogram", HISTOGRAM_RESERVOIR_SIZE); this.deleteFilesSizeHistogram = loadHistogramIfAvailable( - writerMetrics, "deleteFilesSizeHistogram", HISTOGRAM_RESERVOIR_SIZE, classLoader); + writerMetrics, "deleteFilesSizeHistogram", HISTOGRAM_RESERVOIR_SIZE); } public void updateFlushResult(WriteResult result) { @@ -122,48 +157,16 @@ Histogram deleteFilesSizeHistogram() { * Checks whether the Dropwizard-based histogram wrapper provided through Flink's optional * flink-metrics-dropwizard dependency is available. */ - @SuppressWarnings("CatchBlockLogException") private static Histogram loadHistogramIfAvailable( - MetricGroup group, String name, int reservoirSize, ClassLoader classLoader) { - - try { - Class reservoirInterface = - DynClasses.builder() - .loader(classLoader) - .impl("com.codahale.metrics.Reservoir") - .buildChecked(); - Class codahaleHistogramClass = - DynClasses.builder() - .loader(classLoader) - .impl("com.codahale.metrics.Histogram") - .buildChecked(); + MetricGroup group, String name, int reservoirSize) { - Object reservoir = - DynConstructors.builder() - .loader(classLoader) - .impl("com.codahale.metrics.SlidingWindowReservoir", int.class) - .buildChecked() - .newInstance(reservoirSize); - Object codahaleHistogram = - DynConstructors.builder() - .impl(codahaleHistogramClass, reservoirInterface) - .buildChecked() - .newInstance(reservoir); - Histogram wrapper = - DynConstructors.builder(Histogram.class) - .loader(classLoader) - .impl( - "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", - codahaleHistogramClass) - .buildChecked() - .newInstance(codahaleHistogram); - - return group.histogram(name, wrapper); - } catch (ClassNotFoundException | NoSuchMethodException e) { - LOG.warn( - "flink-metrics-dropwizard is not on the classpath. '{}' histogram metrics will be disabled. Add org.apache.flink:flink-metrics-dropwizard to enable them.", - name); + if (!DROPWIZARD_AVAILABLE) { return null; } + + Object reservoir = RESERVOIR_CTOR.newInstance(reservoirSize); + Object codahaleHistogram = CODAHALE_HISTOGRAM_CTOR.newInstance(reservoir); + Histogram wrapper = WRAPPER_CTOR.newInstance(codahaleHistogram); + return group.histogram(name, wrapper); } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java index b50f4269285f..42bbfc0d3628 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java @@ -21,8 +21,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatNoException; -import java.net.URL; -import java.net.URLClassLoader; import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; import org.apache.iceberg.io.WriteResult; import org.junit.jupiter.api.Test; @@ -41,32 +39,4 @@ void histogramsCreatedWhenDropwizardAvailable() { assertThatNoException() .isThrownBy(() -> metrics.updateFlushResult(WriteResult.builder().build())); } - - @Test - void histogramsSkippedWhenDropwizardMissing() throws Exception { - try (EmptyClassloader loader = new EmptyClassloader()) { - - IcebergStreamWriterMetrics metrics = - new IcebergStreamWriterMetrics( - UnregisteredMetricsGroup.createSinkWriterMetricGroup(), "db.table", loader); - - assertThat(metrics.dataFilesSizeHistogram()).isNull(); - assertThat(metrics.deleteFilesSizeHistogram()).isNull(); - - assertThatNoException() - .isThrownBy(() -> metrics.updateFlushResult(WriteResult.builder().build())); - } - } - - private static final class EmptyClassloader extends URLClassLoader { - - EmptyClassloader() { - super(new URL[0]); - } - - @Override - protected Class loadClass(String name, boolean resolve) throws ClassNotFoundException { - throw new ClassNotFoundException(name); - } - } } From 8e99f1b2d99775f5f706386b94c296c3036483cb Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 4 May 2026 05:55:31 +0200 Subject: [PATCH 14/17] fixup! Improve dropwizard-missing warning message --- .../apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index acd56b5c6116..4851110c25b2 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -72,7 +72,7 @@ public class IcebergStreamWriterMetrics { available = true; } catch (ClassNotFoundException | NoSuchMethodException e) { LOG.warn( - "flink-metrics-dropwizard is not on the classpath. Histogram metrics will be disabled. Add org.apache.flink:flink-metrics-dropwizard to enable them.", + "Cannot load Dropwizard metrics; is org.apache.flink:flink-metrics-dropwizard on the classpath?", e); } From 63971c0d09d42b938965414baf60f71f9589d983 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 4 May 2026 05:58:17 +0200 Subject: [PATCH 15/17] fixup! Split histogram construction from group registration --- .../sink/IcebergStreamWriterMetrics.java | 23 ++++++++----------- 1 file changed, 9 insertions(+), 14 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index 4851110c25b2..ee1d1b1997fa 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -98,11 +98,8 @@ public IcebergStreamWriterMetrics(MetricGroup metrics, String fullTableName) { this.lastFlushDurationMs = new AtomicLong(); writerMetrics.gauge("lastFlushDurationMs", lastFlushDurationMs::get); - this.dataFilesSizeHistogram = - loadHistogramIfAvailable(writerMetrics, "dataFilesSizeHistogram", HISTOGRAM_RESERVOIR_SIZE); - this.deleteFilesSizeHistogram = - loadHistogramIfAvailable( - writerMetrics, "deleteFilesSizeHistogram", HISTOGRAM_RESERVOIR_SIZE); + this.dataFilesSizeHistogram = registerHistogram(writerMetrics, "dataFilesSizeHistogram"); + this.deleteFilesSizeHistogram = registerHistogram(writerMetrics, "deleteFilesSizeHistogram"); } public void updateFlushResult(WriteResult result) { @@ -153,20 +150,18 @@ Histogram deleteFilesSizeHistogram() { return deleteFilesSizeHistogram; } - /** - * Checks whether the Dropwizard-based histogram wrapper provided through Flink's optional - * flink-metrics-dropwizard dependency is available. - */ - private static Histogram loadHistogramIfAvailable( - MetricGroup group, String name, int reservoirSize) { + 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_AVAILABLE) { return null; } - Object reservoir = RESERVOIR_CTOR.newInstance(reservoirSize); + Object reservoir = RESERVOIR_CTOR.newInstance(HISTOGRAM_RESERVOIR_SIZE); Object codahaleHistogram = CODAHALE_HISTOGRAM_CTOR.newInstance(reservoir); - Histogram wrapper = WRAPPER_CTOR.newInstance(codahaleHistogram); - return group.histogram(name, wrapper); + return WRAPPER_CTOR.newInstance(codahaleHistogram); } } From b43f620359ee7d5c46b350417321edf369afde50 Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 4 May 2026 06:53:15 +0200 Subject: [PATCH 16/17] fixup! Move fields to a separate class --- .../sink/IcebergStreamWriterMetrics.java | 81 +++++++++---------- 1 file changed, 36 insertions(+), 45 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index ee1d1b1997fa..0563a9a4ffb1 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -40,47 +40,9 @@ public class IcebergStreamWriterMetrics { // 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 - private static final boolean DROPWIZARD_AVAILABLE; - private static final DynConstructors.Ctor RESERVOIR_CTOR; - private static final DynConstructors.Ctor CODAHALE_HISTOGRAM_CTOR; - private static final DynConstructors.Ctor WRAPPER_CTOR; - - static { - DynConstructors.Ctor reservoirCtor = null; - DynConstructors.Ctor codahaleHistogramCtor = null; - DynConstructors.Ctor wrapperCtor = null; - boolean available = false; - - try { - Class reservoirInterface = Class.forName("com.codahale.metrics.Reservoir"); - Class codahaleHistogramClass = Class.forName("com.codahale.metrics.Histogram"); - reservoirCtor = - DynConstructors.builder() - .impl("com.codahale.metrics.SlidingWindowReservoir", int.class) - .buildChecked(); - codahaleHistogramCtor = - DynConstructors.builder() - .impl("com.codahale.metrics.Histogram", reservoirInterface) - .buildChecked(); - wrapperCtor = - DynConstructors.builder(Histogram.class) - .impl( - "org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper", - codahaleHistogramClass) - .buildChecked(); - available = true; - } catch (ClassNotFoundException | NoSuchMethodException e) { - LOG.warn( - "Cannot load Dropwizard metrics; is org.apache.flink:flink-metrics-dropwizard on the classpath?", - e); - } - - RESERVOIR_CTOR = reservoirCtor; - CODAHALE_HISTOGRAM_CTOR = codahaleHistogramCtor; - WRAPPER_CTOR = wrapperCtor; - DROPWIZARD_AVAILABLE = available; - } + // 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; @@ -156,12 +118,41 @@ private static Histogram registerHistogram(MetricGroup group, String name) { } private static Histogram newDropwizardHistogram() { - if (!DROPWIZARD_AVAILABLE) { + if (DROPWIZARD == null) { return null; } - Object reservoir = RESERVOIR_CTOR.newInstance(HISTOGRAM_RESERVOIR_SIZE); - Object codahaleHistogram = CODAHALE_HISTOGRAM_CTOR.newInstance(reservoir); - return WRAPPER_CTOR.newInstance(codahaleHistogram); + 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 = Class.forName("com.codahale.metrics.Reservoir"); + Class codahaleHistogramClass = Class.forName("com.codahale.metrics.Histogram"); + 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 wrapperCtor) {} } From d11e165bebc549f141e40dc5964a9dfe9bc4acbf Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 4 May 2026 12:24:53 +0200 Subject: [PATCH 17/17] fixup! Use DynClasses for all class lookups --- .../iceberg/flink/sink/IcebergStreamWriterMetrics.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index 0563a9a4ffb1..6cf15ff713fb 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java @@ -25,6 +25,7 @@ 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; @@ -129,8 +130,10 @@ private static Histogram newDropwizardHistogram() { private static DropwizardCtors loadDropwizardCtors() { try { - Class reservoirInterface = Class.forName("com.codahale.metrics.Reservoir"); - Class codahaleHistogramClass = Class.forName("com.codahale.metrics.Histogram"); + 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)