Samrat002 commented on code in PR #28427: URL: https://github.com/apache/flink/pull/28427#discussion_r3619885285
########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java: ########## @@ -0,0 +1,281 @@ +/* + * 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.flink.fs.s3native.metrics; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.ThreadSafeSimpleCounter; +import org.apache.flink.util.Preconditions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.http.HttpMetric; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; +import software.amazon.awssdk.metrics.SdkMetric; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Bridges AWS SDK v2's {@link MetricPublisher} into Flink metrics for {@code flink-s3-fs-native}. + * + * <p>The SDK invokes {@link #publish(MetricCollection)} asynchronously after every completed API + * call, on its internal completion executor. The bridge reads a small, fixed set of fields and + * emits the default metric surface of FLIP-576 against the {@code filesystem_type}-labelled scope + * it is handed at construction: + * + * <ul> + * <li>{@code api_call_count} (Counter) — labels {@code op}, {@code status_class} + * <li>{@code api_call_duration_ms} (Histogram) — label {@code op} + * <li>{@code throttle_count} (Counter) — label {@code op} + * <li>{@code retry_count} (Counter) — labels {@code op}, {@code reason} + * </ul> + * + * <p>({@code iops} from the default allowlist is derived at reporter time as the rate of {@code + * api_call_count}, so it is not a separately registered metric.) + * + * <p><b>Allowlist.</b> Only metrics whose name is in the allowlist passed at construction are + * registered; the rest are skipped on the hot path. A {@code "*"} entry registers everything. The + * default set is the five FLIP-576 metrics ({@code api_call_count}, {@code api_call_duration_ms}, + * {@code throttle_count}, {@code retry_count}, {@code iops}). + * + * <p><b>Cardinality.</b> {@code op} comes from the SDK operation name (a closed set of ~15 values + * for S3), {@code status_class} is a closed classifier ({@code 2xx}, {@code 4xx}, {@code 5xx}, + * {@code throttled}, {@code other}, {@code error}, {@code unknown}), and {@code reason} is a closed + * enum ({@code throttled}, {@code 5xx}, {@code other}). Metric handles are cached in bounded maps, + * so {@link #publish} is a map lookup plus a counter increment with no per-record allocation. + * + * <p><b>Thread-safety.</b> Counters use {@link ThreadSafeSimpleCounter} and the histogram is + * synchronized, so concurrent publishes from the SDK completion executor are safe. The bridge is + * also safe to share across multiple S3 clients of the same plugin instance. + */ +@Internal +public final class AwsSdkMetricBridge implements MetricPublisher { + + private static final Logger LOG = LoggerFactory.getLogger(AwsSdkMetricBridge.class); + + static final String API_CALL_COUNT = "api_call_count"; + static final String API_CALL_DURATION_MS = "api_call_duration_ms"; + static final String THROTTLE_COUNT = "throttle_count"; + static final String RETRY_COUNT = "retry_count"; + static final String IOPS = "iops"; + + /** The default-on metric set from FLIP-576. {@code iops} is derived, not registered. */ + static final List<String> DEFAULT_ALLOWLIST = + Arrays.asList(API_CALL_COUNT, API_CALL_DURATION_MS, THROTTLE_COUNT, RETRY_COUNT, IOPS); + + private static final String WILDCARD = "*"; + + private static final String LABEL_OP = "op"; + private static final String LABEL_STATUS_CLASS = "status_class"; + private static final String LABEL_REASON = "reason"; + + private static final String UNKNOWN_OP = "Unknown"; + + private final MetricGroup fsScope; + private final int histogramWindowSize; + + private final boolean allowAll; + private final Set<String> allowlist; + + // op and label sets are closed, so these maps are bounded by construction. + private final ConcurrentHashMap<String, Counter> counters = new ConcurrentHashMap<>(); Review Comment: Done. The shared recorder declares the caches as `Map<...>` with `ConcurrentHashMap` only as the implementation. The AWS adapter no longer owns metric caches after 47babe49e34. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java: ########## @@ -0,0 +1,281 @@ +/* + * 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.flink.fs.s3native.metrics; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.ThreadSafeSimpleCounter; +import org.apache.flink.util.Preconditions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.http.HttpMetric; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; +import software.amazon.awssdk.metrics.SdkMetric; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Bridges AWS SDK v2's {@link MetricPublisher} into Flink metrics for {@code flink-s3-fs-native}. + * + * <p>The SDK invokes {@link #publish(MetricCollection)} asynchronously after every completed API + * call, on its internal completion executor. The bridge reads a small, fixed set of fields and + * emits the default metric surface of FLIP-576 against the {@code filesystem_type}-labelled scope + * it is handed at construction: + * + * <ul> + * <li>{@code api_call_count} (Counter) — labels {@code op}, {@code status_class} + * <li>{@code api_call_duration_ms} (Histogram) — label {@code op} + * <li>{@code throttle_count} (Counter) — label {@code op} + * <li>{@code retry_count} (Counter) — labels {@code op}, {@code reason} + * </ul> + * + * <p>({@code iops} from the default allowlist is derived at reporter time as the rate of {@code + * api_call_count}, so it is not a separately registered metric.) + * + * <p><b>Allowlist.</b> Only metrics whose name is in the allowlist passed at construction are + * registered; the rest are skipped on the hot path. A {@code "*"} entry registers everything. The + * default set is the five FLIP-576 metrics ({@code api_call_count}, {@code api_call_duration_ms}, + * {@code throttle_count}, {@code retry_count}, {@code iops}). + * + * <p><b>Cardinality.</b> {@code op} comes from the SDK operation name (a closed set of ~15 values + * for S3), {@code status_class} is a closed classifier ({@code 2xx}, {@code 4xx}, {@code 5xx}, + * {@code throttled}, {@code other}, {@code error}, {@code unknown}), and {@code reason} is a closed + * enum ({@code throttled}, {@code 5xx}, {@code other}). Metric handles are cached in bounded maps, + * so {@link #publish} is a map lookup plus a counter increment with no per-record allocation. + * + * <p><b>Thread-safety.</b> Counters use {@link ThreadSafeSimpleCounter} and the histogram is + * synchronized, so concurrent publishes from the SDK completion executor are safe. The bridge is + * also safe to share across multiple S3 clients of the same plugin instance. + */ +@Internal +public final class AwsSdkMetricBridge implements MetricPublisher { + + private static final Logger LOG = LoggerFactory.getLogger(AwsSdkMetricBridge.class); + + static final String API_CALL_COUNT = "api_call_count"; + static final String API_CALL_DURATION_MS = "api_call_duration_ms"; + static final String THROTTLE_COUNT = "throttle_count"; + static final String RETRY_COUNT = "retry_count"; + static final String IOPS = "iops"; + + /** The default-on metric set from FLIP-576. {@code iops} is derived, not registered. */ + static final List<String> DEFAULT_ALLOWLIST = + Arrays.asList(API_CALL_COUNT, API_CALL_DURATION_MS, THROTTLE_COUNT, RETRY_COUNT, IOPS); + + private static final String WILDCARD = "*"; + + private static final String LABEL_OP = "op"; + private static final String LABEL_STATUS_CLASS = "status_class"; + private static final String LABEL_REASON = "reason"; + + private static final String UNKNOWN_OP = "Unknown"; + + private final MetricGroup fsScope; + private final int histogramWindowSize; + + private final boolean allowAll; + private final Set<String> allowlist; + + // op and label sets are closed, so these maps are bounded by construction. + private final ConcurrentHashMap<String, Counter> counters = new ConcurrentHashMap<>(); + private final ConcurrentHashMap<String, S3MetricHistogram> histograms = + new ConcurrentHashMap<>(); + + public AwsSdkMetricBridge(MetricGroup fsScope) { + this(fsScope, DEFAULT_ALLOWLIST, S3MetricHistogram.DEFAULT_WINDOW_SIZE); + } + + public AwsSdkMetricBridge(MetricGroup fsScope, int histogramWindowSize) { + this(fsScope, DEFAULT_ALLOWLIST, histogramWindowSize); + } + + public AwsSdkMetricBridge( + MetricGroup fsScope, @Nullable Collection<String> allowlist, int histogramWindowSize) { + this.fsScope = Preconditions.checkNotNull(fsScope, "fsScope must not be null"); + Preconditions.checkArgument( + histogramWindowSize > 0, "histogramWindowSize must be positive"); + this.histogramWindowSize = histogramWindowSize; + + if (allowlist == null || allowlist.isEmpty()) { Review Comment: Addressed in 47babe49e34. Defaults are owned by the scheme-parameterized `FileSystemMetricOptions`; an explicitly empty list is rejected rather than falling back. The universal option shape and allowlist handling moved to `flink-core`, while only the concrete `s3` prefix remains in the native factory. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java: ########## @@ -0,0 +1,281 @@ +/* + * 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.flink.fs.s3native.metrics; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.ThreadSafeSimpleCounter; +import org.apache.flink.util.Preconditions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.http.HttpMetric; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; +import software.amazon.awssdk.metrics.SdkMetric; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Bridges AWS SDK v2's {@link MetricPublisher} into Flink metrics for {@code flink-s3-fs-native}. + * + * <p>The SDK invokes {@link #publish(MetricCollection)} asynchronously after every completed API + * call, on its internal completion executor. The bridge reads a small, fixed set of fields and + * emits the default metric surface of FLIP-576 against the {@code filesystem_type}-labelled scope + * it is handed at construction: + * + * <ul> + * <li>{@code api_call_count} (Counter) — labels {@code op}, {@code status_class} + * <li>{@code api_call_duration_ms} (Histogram) — label {@code op} + * <li>{@code throttle_count} (Counter) — label {@code op} + * <li>{@code retry_count} (Counter) — labels {@code op}, {@code reason} + * </ul> + * + * <p>({@code iops} from the default allowlist is derived at reporter time as the rate of {@code + * api_call_count}, so it is not a separately registered metric.) + * + * <p><b>Allowlist.</b> Only metrics whose name is in the allowlist passed at construction are + * registered; the rest are skipped on the hot path. A {@code "*"} entry registers everything. The + * default set is the five FLIP-576 metrics ({@code api_call_count}, {@code api_call_duration_ms}, + * {@code throttle_count}, {@code retry_count}, {@code iops}). + * + * <p><b>Cardinality.</b> {@code op} comes from the SDK operation name (a closed set of ~15 values + * for S3), {@code status_class} is a closed classifier ({@code 2xx}, {@code 4xx}, {@code 5xx}, + * {@code throttled}, {@code other}, {@code error}, {@code unknown}), and {@code reason} is a closed + * enum ({@code throttled}, {@code 5xx}, {@code other}). Metric handles are cached in bounded maps, + * so {@link #publish} is a map lookup plus a counter increment with no per-record allocation. + * + * <p><b>Thread-safety.</b> Counters use {@link ThreadSafeSimpleCounter} and the histogram is + * synchronized, so concurrent publishes from the SDK completion executor are safe. The bridge is + * also safe to share across multiple S3 clients of the same plugin instance. + */ +@Internal +public final class AwsSdkMetricBridge implements MetricPublisher { + + private static final Logger LOG = LoggerFactory.getLogger(AwsSdkMetricBridge.class); + + static final String API_CALL_COUNT = "api_call_count"; + static final String API_CALL_DURATION_MS = "api_call_duration_ms"; + static final String THROTTLE_COUNT = "throttle_count"; + static final String RETRY_COUNT = "retry_count"; + static final String IOPS = "iops"; + + /** The default-on metric set from FLIP-576. {@code iops} is derived, not registered. */ + static final List<String> DEFAULT_ALLOWLIST = + Arrays.asList(API_CALL_COUNT, API_CALL_DURATION_MS, THROTTLE_COUNT, RETRY_COUNT, IOPS); + + private static final String WILDCARD = "*"; + + private static final String LABEL_OP = "op"; + private static final String LABEL_STATUS_CLASS = "status_class"; + private static final String LABEL_REASON = "reason"; + + private static final String UNKNOWN_OP = "Unknown"; + + private final MetricGroup fsScope; + private final int histogramWindowSize; + + private final boolean allowAll; + private final Set<String> allowlist; + + // op and label sets are closed, so these maps are bounded by construction. + private final ConcurrentHashMap<String, Counter> counters = new ConcurrentHashMap<>(); + private final ConcurrentHashMap<String, S3MetricHistogram> histograms = + new ConcurrentHashMap<>(); + + public AwsSdkMetricBridge(MetricGroup fsScope) { + this(fsScope, DEFAULT_ALLOWLIST, S3MetricHistogram.DEFAULT_WINDOW_SIZE); + } + + public AwsSdkMetricBridge(MetricGroup fsScope, int histogramWindowSize) { + this(fsScope, DEFAULT_ALLOWLIST, histogramWindowSize); + } + + public AwsSdkMetricBridge( + MetricGroup fsScope, @Nullable Collection<String> allowlist, int histogramWindowSize) { + this.fsScope = Preconditions.checkNotNull(fsScope, "fsScope must not be null"); + Preconditions.checkArgument( + histogramWindowSize > 0, "histogramWindowSize must be positive"); + this.histogramWindowSize = histogramWindowSize; + + if (allowlist == null || allowlist.isEmpty()) { + LOG.warn( + "S3 metrics allowlist is empty; falling back to the default metric set {}", + DEFAULT_ALLOWLIST); + this.allowAll = false; + this.allowlist = new HashSet<>(DEFAULT_ALLOWLIST); + } else if (allowlist.contains(WILDCARD)) { + this.allowAll = true; + this.allowlist = new HashSet<>(); + } else { + this.allowAll = false; + this.allowlist = new HashSet<>(allowlist); + } + } + + private boolean allowed(String metricName) { + return allowAll || allowlist.contains(metricName); + } + + @Override + public void publish(MetricCollection apiCall) { + try { + translate(apiCall); + } catch (Throwable t) { Review Comment: No. This now catches `Exception`, so JVM errors such as OOM are not swallowed. ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridgeTest.java: ########## @@ -0,0 +1,231 @@ +/* + * 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.flink.fs.s3native.metrics; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; + +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.http.HttpMetric; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricCollector; + +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link AwsSdkMetricBridge}'s translation of SDK metric records into Flink metrics. */ +class AwsSdkMetricBridgeTest { + + @Test + void successfulCallIncrementsApiCallCountAndRecordsDuration() { + CapturingMetricGroup root = new CapturingMetricGroup(); + AwsSdkMetricBridge bridge = new AwsSdkMetricBridge(root); + + bridge.publish(apiCall("PutObject", Duration.ofMillis(120), true, 0, 200)); + + assertThat(root.count("op=PutObject/status_class=2xx/api_call_count")).isEqualTo(1L); + Histogram histogram = root.histograms.get("op=PutObject/api_call_duration_ms"); + assertThat(histogram).isNotNull(); + assertThat(histogram.getCount()).isEqualTo(1L); + assertThat(histogram.getStatistics().getMax()).isEqualTo(120L); + assertThat(root.counters).doesNotContainKey("op=PutObject/throttle_count"); + } + + @Test + void throttledCallIncrementsThrottleAndRetryCounts() { + CapturingMetricGroup root = new CapturingMetricGroup(); + AwsSdkMetricBridge bridge = new AwsSdkMetricBridge(root); + + // Two throttled attempts (503) followed by a successful one (200); RETRY_COUNT = 2. + bridge.publish(apiCall("UploadPart", Duration.ofMillis(900), true, 2, 503, 503, 200)); + + assertThat(root.count("op=UploadPart/throttle_count")).isEqualTo(2L); + assertThat(root.count("op=UploadPart/reason=throttled/retry_count")).isEqualTo(2L); + // The final attempt succeeded, so the overall call is classified 2xx. + assertThat(root.count("op=UploadPart/status_class=2xx/api_call_count")).isEqualTo(1L); + } + + @Test + void clientErrorIsClassifiedAs4xx() { + CapturingMetricGroup root = new CapturingMetricGroup(); + AwsSdkMetricBridge bridge = new AwsSdkMetricBridge(root); + + bridge.publish(apiCall("HeadObject", Duration.ofMillis(20), false, 0, 404)); + + assertThat(root.count("op=HeadObject/status_class=4xx/api_call_count")).isEqualTo(1L); + assertThat(root.counters).doesNotContainKey("op=HeadObject/throttle_count"); + } + + @Test + void allowlistRegistersOnlyTheListedMetrics() { + CapturingMetricGroup root = new CapturingMetricGroup(); + // Only api_call_count is allowed; duration, throttle and retry must be skipped. + AwsSdkMetricBridge bridge = + new AwsSdkMetricBridge( + root, + Collections.singletonList(AwsSdkMetricBridge.API_CALL_COUNT), + S3MetricHistogram.DEFAULT_WINDOW_SIZE); + + bridge.publish(apiCall("UploadPart", Duration.ofMillis(900), true, 2, 503, 503, 200)); + + assertThat(root.count("op=UploadPart/status_class=2xx/api_call_count")).isEqualTo(1L); + assertThat(root.histograms).doesNotContainKey("op=UploadPart/api_call_duration_ms"); + assertThat(root.counters).doesNotContainKey("op=UploadPart/throttle_count"); + assertThat(root.counters).doesNotContainKey("op=UploadPart/reason=throttled/retry_count"); + } + + @Test + void wildcardAllowlistRegistersEveryMetric() { Review Comment: Good point. I replaced the output-based wildcard test in 288c21d7efa with a direct shared-recorder test: `*` enables a synthetic future/non-default metric, while `DEFAULT_ALLOWLIST` does not. That now distinguishes wildcard behavior from default fallback. ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridgeTest.java: ########## @@ -0,0 +1,231 @@ +/* + * 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.flink.fs.s3native.metrics; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; + +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.http.HttpMetric; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricCollector; + +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link AwsSdkMetricBridge}'s translation of SDK metric records into Flink metrics. */ +class AwsSdkMetricBridgeTest { + + @Test + void successfulCallIncrementsApiCallCountAndRecordsDuration() { + CapturingMetricGroup root = new CapturingMetricGroup(); + AwsSdkMetricBridge bridge = new AwsSdkMetricBridge(root); + + bridge.publish(apiCall("PutObject", Duration.ofMillis(120), true, 0, 200)); + + assertThat(root.count("op=PutObject/status_class=2xx/api_call_count")).isEqualTo(1L); + Histogram histogram = root.histograms.get("op=PutObject/api_call_duration_ms"); + assertThat(histogram).isNotNull(); + assertThat(histogram.getCount()).isEqualTo(1L); + assertThat(histogram.getStatistics().getMax()).isEqualTo(120L); + assertThat(root.counters).doesNotContainKey("op=PutObject/throttle_count"); + } + + @Test + void throttledCallIncrementsThrottleAndRetryCounts() { + CapturingMetricGroup root = new CapturingMetricGroup(); + AwsSdkMetricBridge bridge = new AwsSdkMetricBridge(root); + + // Two throttled attempts (503) followed by a successful one (200); RETRY_COUNT = 2. + bridge.publish(apiCall("UploadPart", Duration.ofMillis(900), true, 2, 503, 503, 200)); + + assertThat(root.count("op=UploadPart/throttle_count")).isEqualTo(2L); + assertThat(root.count("op=UploadPart/reason=throttled/retry_count")).isEqualTo(2L); + // The final attempt succeeded, so the overall call is classified 2xx. + assertThat(root.count("op=UploadPart/status_class=2xx/api_call_count")).isEqualTo(1L); + } + + @Test + void clientErrorIsClassifiedAs4xx() { + CapturingMetricGroup root = new CapturingMetricGroup(); + AwsSdkMetricBridge bridge = new AwsSdkMetricBridge(root); + + bridge.publish(apiCall("HeadObject", Duration.ofMillis(20), false, 0, 404)); + + assertThat(root.count("op=HeadObject/status_class=4xx/api_call_count")).isEqualTo(1L); + assertThat(root.counters).doesNotContainKey("op=HeadObject/throttle_count"); + } + + @Test + void allowlistRegistersOnlyTheListedMetrics() { + CapturingMetricGroup root = new CapturingMetricGroup(); + // Only api_call_count is allowed; duration, throttle and retry must be skipped. + AwsSdkMetricBridge bridge = + new AwsSdkMetricBridge( + root, + Collections.singletonList(AwsSdkMetricBridge.API_CALL_COUNT), + S3MetricHistogram.DEFAULT_WINDOW_SIZE); + + bridge.publish(apiCall("UploadPart", Duration.ofMillis(900), true, 2, 503, 503, 200)); + + assertThat(root.count("op=UploadPart/status_class=2xx/api_call_count")).isEqualTo(1L); + assertThat(root.histograms).doesNotContainKey("op=UploadPart/api_call_duration_ms"); + assertThat(root.counters).doesNotContainKey("op=UploadPart/throttle_count"); + assertThat(root.counters).doesNotContainKey("op=UploadPart/reason=throttled/retry_count"); + } + + @Test + void wildcardAllowlistRegistersEveryMetric() { + CapturingMetricGroup root = new CapturingMetricGroup(); + AwsSdkMetricBridge bridge = + new AwsSdkMetricBridge( + root, + Collections.singletonList("*"), + S3MetricHistogram.DEFAULT_WINDOW_SIZE); + + bridge.publish(apiCall("UploadPart", Duration.ofMillis(900), true, 2, 503, 503, 200)); + + assertThat(root.count("op=UploadPart/status_class=2xx/api_call_count")).isEqualTo(1L); + assertThat(root.histograms.get("op=UploadPart/api_call_duration_ms")).isNotNull(); + assertThat(root.count("op=UploadPart/throttle_count")).isEqualTo(2L); + assertThat(root.count("op=UploadPart/reason=throttled/retry_count")).isEqualTo(2L); + } + + @Test + void emptyAllowlistFallsBackToDefaults() { Review Comment: Agreed. An empty allowlist now throws `IllegalArgumentException` during recorder creation; disabling all metrics remains the responsibility of `s3.metrics.enabled=false`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
