From db4cb4f8178b1d327ffdf0b47117529eaba7d653 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:57:34 +0000 Subject: [PATCH 1/5] Initial plan From 5ff0093679bd304d829fffb7795479dbbfe934fa Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 12 Oct 2026 15:07:56 +0000 Subject: [PATCH 2/5] Reduce lock scope in Kafka metrics reporter Co-authored-by: trask <218610+trask@users.noreply.github.com> --- .../OpenTelemetryMetricsReporter.java | 58 +++++++++++++------ 1 file changed, 41 insertions(+), 17 deletions(-) diff --git a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java index 440f02e843ba..fa7126badc27 100644 --- a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java +++ b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java @@ -98,29 +98,36 @@ public void metricChange(KafkaMetric metric) { } Set> attributeKeys = registeredObservable.getAttributes().asMap().keySet(); + List observablesToClose = new ArrayList<>(); synchronized (lock) { - for (Iterator it = registeredObservables.iterator(); it.hasNext(); ) { - RegisteredObservable curRegisteredObservable = it.next(); + for (RegisteredObservable curRegisteredObservable : registeredObservables) { Set> curAttributeKeys = curRegisteredObservable.getAttributes().asMap().keySet(); if (curRegisteredObservable.getKafkaMetricName().equals(metric.metricName())) { - logger.log(FINEST, "Replacing instrument: {0}", curRegisteredObservable); - closeInstrument(curRegisteredObservable.getObservable()); - it.remove(); + observablesToClose.add(curRegisteredObservable); } else if (curRegisteredObservable .getInstrumentDescriptor() .equals(registeredObservable.getInstrumentDescriptor()) && attributeKeys.size() > curAttributeKeys.size() && attributeKeys.containsAll(curAttributeKeys)) { - logger.log( - FINEST, - "Replacing instrument with higher dimension version: {0}", - curRegisteredObservable); - closeInstrument(curRegisteredObservable.getObservable()); - it.remove(); + observablesToClose.add(curRegisteredObservable); } } + } + + for (RegisteredObservable curRegisteredObservable : observablesToClose) { + if (curRegisteredObservable.getKafkaMetricName().equals(metric.metricName())) { + logger.log(FINEST, "Replacing instrument: {0}", curRegisteredObservable); + } else { + logger.log( + FINEST, + "Replacing instrument with higher dimension version: {0}", + curRegisteredObservable); + } + closeInstrumentAndRemove(curRegisteredObservable); + } + synchronized (lock) { registeredObservables.add(registeredObservable); } } @@ -128,15 +135,18 @@ public void metricChange(KafkaMetric metric) { @Override public void metricRemoval(KafkaMetric metric) { logger.log(FINEST, "Metric removed: {0}", metric.metricName()); + List observablesToClose = new ArrayList<>(); synchronized (lock) { - for (Iterator it = registeredObservables.iterator(); it.hasNext(); ) { - RegisteredObservable current = it.next(); + for (RegisteredObservable current : registeredObservables) { if (current.getKafkaMetricName().equals(metric.metricName())) { - closeInstrument(current.getObservable()); - it.remove(); + observablesToClose.add(current); } } } + + for (RegisteredObservable current : observablesToClose) { + closeInstrumentAndRemove(current); + } } @Override @@ -163,10 +173,24 @@ public void onUpdate(ClusterResource clusterResource) { } private void closeAllInstruments() { + List observablesToClose; + synchronized (lock) { + observablesToClose = new ArrayList<>(registeredObservables); + } + + for (RegisteredObservable registeredObservable : observablesToClose) { + closeInstrumentAndRemove(registeredObservable); + } + } + + private void closeInstrumentAndRemove(RegisteredObservable registeredObservable) { + closeInstrument(registeredObservable.getObservable()); synchronized (lock) { for (Iterator it = registeredObservables.iterator(); it.hasNext(); ) { - closeInstrument(it.next().getObservable()); - it.remove(); + if (it.next() == registeredObservable) { + it.remove(); + break; + } } } } From 47f5e2b6e16fa562a2bea4dc5e6ce25fd4c12fd1 Mon Sep 17 00:00:00 2001 From: Trask Stalnaker Date: Tue, 13 Oct 2026 08:36:56 -0700 Subject: [PATCH 3/5] Simplify Kafka observable registration removal Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: ee62c0a6-39b4-4534-92d6-83f191bb8a8c --- .../v0_11/internal/OpenTelemetryMetricsReporter.java | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java index fa7126badc27..f8bf4364f1b4 100644 --- a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java +++ b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java @@ -14,7 +14,6 @@ import io.opentelemetry.instrumentation.api.internal.EmbeddedInstrumentationProperties; import io.opentelemetry.instrumentation.api.internal.GuardedBy; import java.util.ArrayList; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; @@ -186,12 +185,7 @@ private void closeAllInstruments() { private void closeInstrumentAndRemove(RegisteredObservable registeredObservable) { closeInstrument(registeredObservable.getObservable()); synchronized (lock) { - for (Iterator it = registeredObservables.iterator(); it.hasNext(); ) { - if (it.next() == registeredObservable) { - it.remove(); - break; - } - } + registeredObservables.remove(registeredObservable); } } From 5b2df7c175b1709ea358aebb35703d4ff2cd49a4 Mon Sep 17 00:00:00 2001 From: Trask Stalnaker Date: Tue, 13 Oct 2026 08:42:59 -0700 Subject: [PATCH 4/5] Use ordered set for Kafka observable registrations Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: ee62c0a6-39b4-4534-92d6-83f191bb8a8c --- .../OpenTelemetryMetricsReporter.java | 3 +- .../OpenTelemetryMetricsReporterTest.java | 186 ++++++++++++++++++ 2 files changed, 188 insertions(+), 1 deletion(-) create mode 100644 instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporterTest.java diff --git a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java index f8bf4364f1b4..92e76a9d049c 100644 --- a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java +++ b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporter.java @@ -14,6 +14,7 @@ import io.opentelemetry.instrumentation.api.internal.EmbeddedInstrumentationProperties; import io.opentelemetry.instrumentation.api.internal.GuardedBy; import java.util.ArrayList; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -53,7 +54,7 @@ public final class OpenTelemetryMetricsReporter private final Object lock = new Object(); @GuardedBy("lock") - private final List registeredObservables = new ArrayList<>(); + private final Set registeredObservables = new LinkedHashSet<>(); /** * Reset for test by resetting the {@link #meter} to {@code null} and closing all registered diff --git a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporterTest.java b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporterTest.java new file mode 100644 index 000000000000..a35ce6bb8edd --- /dev/null +++ b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporterTest.java @@ -0,0 +1,186 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal; + +import static java.util.Collections.emptyMap; +import static java.util.Collections.singletonMap; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.metrics.ObservableDoubleGauge; +import java.util.HashMap; +import java.util.Map; +import org.apache.kafka.common.MetricName; +import org.apache.kafka.common.metrics.KafkaMetric; +import org.apache.kafka.common.metrics.stats.Avg; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.InOrder; + +class OpenTelemetryMetricsReporterTest { + + private final Meter meter = mock(Meter.class, RETURNS_DEEP_STUBS); + private final OpenTelemetryMetricsReporter reporter = new OpenTelemetryMetricsReporter(); + + @BeforeEach + void configure() { + OpenTelemetry openTelemetry = mock(OpenTelemetry.class, RETURNS_DEEP_STUBS); + when(openTelemetry.meterBuilder("test").build()).thenReturn(meter); + Map config = new HashMap<>(); + config.put( + OpenTelemetryMetricsReporter.CONFIG_KEY_OPENTELEMETRY_SUPPLIER, + new OpenTelemetrySupplier(openTelemetry)); + config.put(OpenTelemetryMetricsReporter.CONFIG_KEY_OPENTELEMETRY_INSTRUMENTATION_NAME, "test"); + reporter.configure(config); + } + + @Test + void closePreservesRegistrationOrderAndRemovesEachSuccessfulClose() { + ObservableDoubleGauge first = mock(ObservableDoubleGauge.class); + ObservableDoubleGauge second = mock(ObservableDoubleGauge.class); + ObservableDoubleGauge third = mock(ObservableDoubleGauge.class); + registerMetric("first", emptyMap(), first); + registerMetric("second", emptyMap(), second); + registerMetric("third", emptyMap(), third); + doAnswer( + invocation -> { + assertThat(reporter.getRegisteredObservables()) + .extracting(RegisteredObservable::getObservable) + .containsExactly(second, third); + return null; + }) + .when(second) + .close(); + + reporter.close(); + + InOrder order = inOrder(first, second, third); + order.verify(first).close(); + order.verify(second).close(); + order.verify(third).close(); + assertThat(reporter.getRegisteredObservables()).isEmpty(); + + reporter.close(); + + verify(first).close(); + verify(second).close(); + verify(third).close(); + } + + @Test + void failedCloseRetainsFailedAndUnattemptedRegistrationsForRetry() { + ObservableDoubleGauge first = mock(ObservableDoubleGauge.class); + ObservableDoubleGauge second = mock(ObservableDoubleGauge.class); + ObservableDoubleGauge third = mock(ObservableDoubleGauge.class); + registerMetric("first", emptyMap(), first); + registerMetric("second", emptyMap(), second); + registerMetric("third", emptyMap(), third); + IllegalStateException failure = new IllegalStateException("close failed"); + doThrow(failure).doNothing().when(second).close(); + + assertThatThrownBy(reporter::close) + .isInstanceOf(IllegalStateException.class) + .hasMessage("Error occurred closing instrument") + .hasCause(failure); + verify(first).close(); + verify(second).close(); + verify(third, never()).close(); + assertThat(reporter.getRegisteredObservables()) + .extracting(RegisteredObservable::getObservable) + .containsExactly(second, third); + + reporter.close(); + + verify(first).close(); + verify(second, times(2)).close(); + verify(third).close(); + assertThat(reporter.getRegisteredObservables()).isEmpty(); + } + + @Test + void replacementMovesRegistrationToEndAndRemovalClosesOnlyMatchingMetric() { + ObservableDoubleGauge first = mock(ObservableDoubleGauge.class); + ObservableDoubleGauge second = mock(ObservableDoubleGauge.class); + ObservableDoubleGauge replacement = mock(ObservableDoubleGauge.class); + KafkaMetric firstMetric = registerMetric("first", emptyMap(), first); + registerMetric("second", emptyMap(), second); + + registerMetric("first", emptyMap(), replacement); + + verify(first).close(); + assertThat(reporter.getRegisteredObservables()) + .extracting(RegisteredObservable::getObservable) + .containsExactly(second, replacement); + + reporter.metricRemoval(firstMetric); + + verify(replacement).close(); + assertThat(reporter.getRegisteredObservables()) + .extracting(RegisteredObservable::getObservable) + .containsExactly(second); + + reporter.close(); + + InOrder order = inOrder(first, replacement, second); + order.verify(first).close(); + order.verify(replacement).close(); + order.verify(second).close(); + assertThat(reporter.getRegisteredObservables()).isEmpty(); + } + + @Test + void higherDimensionReplacementClosesRegistrationsInOrder() { + ObservableDoubleGauge first = mock(ObservableDoubleGauge.class); + ObservableDoubleGauge second = mock(ObservableDoubleGauge.class); + ObservableDoubleGauge replacement = mock(ObservableDoubleGauge.class); + registerMetric("records", singletonMap("topic", "test-topic"), first); + registerMetric("records", singletonMap("client-id", "test-client"), second); + Map tags = new HashMap<>(); + tags.put("topic", "test-topic"); + tags.put("client-id", "test-client"); + + registerMetric("records", tags, replacement); + + InOrder order = inOrder(first, second, replacement); + order.verify(first).close(); + order.verify(second).close(); + assertThat(reporter.getRegisteredObservables()) + .extracting(RegisteredObservable::getObservable) + .containsExactly(replacement); + + reporter.close(); + + order.verify(replacement).close(); + assertThat(reporter.getRegisteredObservables()).isEmpty(); + } + + private KafkaMetric registerMetric( + String name, Map tags, ObservableDoubleGauge observable) { + when(meter + .gaugeBuilder("kafka.producer." + name) + .setDescription("description") + .buildWithCallback(any())) + .thenReturn(observable); + KafkaMetric metric = mock(KafkaMetric.class); + when(metric.metricName()) + .thenReturn(new MetricName(name, "producer-metrics", "description", tags)); + when(metric.measurable()).thenReturn(new Avg()); + reporter.metricChange(metric); + return metric; + } +} From d16a281aebdb5b0ed9305a4e9ab4cbd1619c013a Mon Sep 17 00:00:00 2001 From: Trask Stalnaker Date: Tue, 13 Oct 2026 09:03:02 -0700 Subject: [PATCH 5/5] Remove new Kafka metrics reporter tests Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0c244542-4785-4886-a331-68c25ca59fa2 --- .../OpenTelemetryMetricsReporterTest.java | 186 ------------------ 1 file changed, 186 deletions(-) delete mode 100644 instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporterTest.java diff --git a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporterTest.java b/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporterTest.java deleted file mode 100644 index a35ce6bb8edd..000000000000 --- a/instrumentation/kafka/kafka-clients/kafka-clients-common-0.11/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/common/v0_11/internal/OpenTelemetryMetricsReporterTest.java +++ /dev/null @@ -1,186 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - */ - -package io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal; - -import static java.util.Collections.emptyMap; -import static java.util.Collections.singletonMap; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.RETURNS_DEEP_STUBS; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.doThrow; -import static org.mockito.Mockito.inOrder; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -import io.opentelemetry.api.OpenTelemetry; -import io.opentelemetry.api.metrics.Meter; -import io.opentelemetry.api.metrics.ObservableDoubleGauge; -import java.util.HashMap; -import java.util.Map; -import org.apache.kafka.common.MetricName; -import org.apache.kafka.common.metrics.KafkaMetric; -import org.apache.kafka.common.metrics.stats.Avg; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.mockito.InOrder; - -class OpenTelemetryMetricsReporterTest { - - private final Meter meter = mock(Meter.class, RETURNS_DEEP_STUBS); - private final OpenTelemetryMetricsReporter reporter = new OpenTelemetryMetricsReporter(); - - @BeforeEach - void configure() { - OpenTelemetry openTelemetry = mock(OpenTelemetry.class, RETURNS_DEEP_STUBS); - when(openTelemetry.meterBuilder("test").build()).thenReturn(meter); - Map config = new HashMap<>(); - config.put( - OpenTelemetryMetricsReporter.CONFIG_KEY_OPENTELEMETRY_SUPPLIER, - new OpenTelemetrySupplier(openTelemetry)); - config.put(OpenTelemetryMetricsReporter.CONFIG_KEY_OPENTELEMETRY_INSTRUMENTATION_NAME, "test"); - reporter.configure(config); - } - - @Test - void closePreservesRegistrationOrderAndRemovesEachSuccessfulClose() { - ObservableDoubleGauge first = mock(ObservableDoubleGauge.class); - ObservableDoubleGauge second = mock(ObservableDoubleGauge.class); - ObservableDoubleGauge third = mock(ObservableDoubleGauge.class); - registerMetric("first", emptyMap(), first); - registerMetric("second", emptyMap(), second); - registerMetric("third", emptyMap(), third); - doAnswer( - invocation -> { - assertThat(reporter.getRegisteredObservables()) - .extracting(RegisteredObservable::getObservable) - .containsExactly(second, third); - return null; - }) - .when(second) - .close(); - - reporter.close(); - - InOrder order = inOrder(first, second, third); - order.verify(first).close(); - order.verify(second).close(); - order.verify(third).close(); - assertThat(reporter.getRegisteredObservables()).isEmpty(); - - reporter.close(); - - verify(first).close(); - verify(second).close(); - verify(third).close(); - } - - @Test - void failedCloseRetainsFailedAndUnattemptedRegistrationsForRetry() { - ObservableDoubleGauge first = mock(ObservableDoubleGauge.class); - ObservableDoubleGauge second = mock(ObservableDoubleGauge.class); - ObservableDoubleGauge third = mock(ObservableDoubleGauge.class); - registerMetric("first", emptyMap(), first); - registerMetric("second", emptyMap(), second); - registerMetric("third", emptyMap(), third); - IllegalStateException failure = new IllegalStateException("close failed"); - doThrow(failure).doNothing().when(second).close(); - - assertThatThrownBy(reporter::close) - .isInstanceOf(IllegalStateException.class) - .hasMessage("Error occurred closing instrument") - .hasCause(failure); - verify(first).close(); - verify(second).close(); - verify(third, never()).close(); - assertThat(reporter.getRegisteredObservables()) - .extracting(RegisteredObservable::getObservable) - .containsExactly(second, third); - - reporter.close(); - - verify(first).close(); - verify(second, times(2)).close(); - verify(third).close(); - assertThat(reporter.getRegisteredObservables()).isEmpty(); - } - - @Test - void replacementMovesRegistrationToEndAndRemovalClosesOnlyMatchingMetric() { - ObservableDoubleGauge first = mock(ObservableDoubleGauge.class); - ObservableDoubleGauge second = mock(ObservableDoubleGauge.class); - ObservableDoubleGauge replacement = mock(ObservableDoubleGauge.class); - KafkaMetric firstMetric = registerMetric("first", emptyMap(), first); - registerMetric("second", emptyMap(), second); - - registerMetric("first", emptyMap(), replacement); - - verify(first).close(); - assertThat(reporter.getRegisteredObservables()) - .extracting(RegisteredObservable::getObservable) - .containsExactly(second, replacement); - - reporter.metricRemoval(firstMetric); - - verify(replacement).close(); - assertThat(reporter.getRegisteredObservables()) - .extracting(RegisteredObservable::getObservable) - .containsExactly(second); - - reporter.close(); - - InOrder order = inOrder(first, replacement, second); - order.verify(first).close(); - order.verify(replacement).close(); - order.verify(second).close(); - assertThat(reporter.getRegisteredObservables()).isEmpty(); - } - - @Test - void higherDimensionReplacementClosesRegistrationsInOrder() { - ObservableDoubleGauge first = mock(ObservableDoubleGauge.class); - ObservableDoubleGauge second = mock(ObservableDoubleGauge.class); - ObservableDoubleGauge replacement = mock(ObservableDoubleGauge.class); - registerMetric("records", singletonMap("topic", "test-topic"), first); - registerMetric("records", singletonMap("client-id", "test-client"), second); - Map tags = new HashMap<>(); - tags.put("topic", "test-topic"); - tags.put("client-id", "test-client"); - - registerMetric("records", tags, replacement); - - InOrder order = inOrder(first, second, replacement); - order.verify(first).close(); - order.verify(second).close(); - assertThat(reporter.getRegisteredObservables()) - .extracting(RegisteredObservable::getObservable) - .containsExactly(replacement); - - reporter.close(); - - order.verify(replacement).close(); - assertThat(reporter.getRegisteredObservables()).isEmpty(); - } - - private KafkaMetric registerMetric( - String name, Map tags, ObservableDoubleGauge observable) { - when(meter - .gaugeBuilder("kafka.producer." + name) - .setDescription("description") - .buildWithCallback(any())) - .thenReturn(observable); - KafkaMetric metric = mock(KafkaMetric.class); - when(metric.metricName()) - .thenReturn(new MetricName(name, "producer-metrics", "description", tags)); - when(metric.measurable()).thenReturn(new Avg()); - reporter.metricChange(metric); - return metric; - } -}