diff --git a/flink/v1.20/build.gradle b/flink/v1.20/build.gradle index 2bbad1891c81..c7ca24817bc9 100644 --- a/flink/v1.20/build.gradle +++ b/flink/v1.20/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.flink120.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/v1.20/flink-runtime/LICENSE b/flink/v1.20/flink-runtime/LICENSE index 11460c3307c8..364652a5aca2 100644 --- a/flink/v1.20/flink-runtime/LICENSE +++ b/flink/v1.20/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/v1.20/flink-runtime/runtime-deps.txt b/flink/v1.20/flink-runtime/runtime-deps.txt index 7c7aed1e4357..00c53ed388d0 100644 --- a/flink/v1.20/flink-runtime/runtime-deps.txt +++ b/flink/v1.20/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:1.20.1 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/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index 434f3969577f..6cf15ff713fb 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v1.20/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 wrapperCtor) {} } diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java new file mode 100644 index 000000000000..42bbfc0d3628 --- /dev/null +++ b/flink/v1.20/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())); + } +} diff --git a/flink/v2.0/build.gradle b/flink/v2.0/build.gradle index 626cc01b28e3..94f851e03221 100644 --- a/flink/v2.0/build.gradle +++ b/flink/v2.0/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.flink20.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.0/flink-runtime/LICENSE b/flink/v2.0/flink-runtime/LICENSE index 11460c3307c8..364652a5aca2 100644 --- a/flink/v2.0/flink-runtime/LICENSE +++ b/flink/v2.0/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.0/flink-runtime/runtime-deps.txt b/flink/v2.0/flink-runtime/runtime-deps.txt index c70e3fbba92c..00c53ed388d0 100644 --- a/flink/v2.0/flink-runtime/runtime-deps.txt +++ b/flink/v2.0/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.0.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.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java index 434f3969577f..6cf15ff713fb 100644 --- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriterMetrics.java +++ b/flink/v2.0/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 wrapperCtor) {} } diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriterMetrics.java new file mode 100644 index 000000000000..42bbfc0d3628 --- /dev/null +++ b/flink/v2.0/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())); + } +}