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..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,7 +14,7 @@
 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.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -54,7 +54,7 @@ public final class OpenTelemetryMetricsReporter
   private final Object lock = new Object();
 
   @GuardedBy("lock")
-  private final List<RegisteredObservable> registeredObservables = new ArrayList<>();
+  private final Set<RegisteredObservable> registeredObservables = new LinkedHashSet<>();
 
   /**
    * Reset for test by resetting the {@link #meter} to {@code null} and closing all registered
@@ -98,29 +98,36 @@ public void metricChange(KafkaMetric metric) {
     }
 
     Set<AttributeKey<?>> attributeKeys = registeredObservable.getAttributes().asMap().keySet();
+    List<RegisteredObservable> observablesToClose = new ArrayList<>();
     synchronized (lock) {
-      for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
-        RegisteredObservable curRegisteredObservable = it.next();
+      for (RegisteredObservable curRegisteredObservable : registeredObservables) {
         Set<AttributeKey<?>> 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<RegisteredObservable> observablesToClose = new ArrayList<>();
     synchronized (lock) {
-      for (Iterator<RegisteredObservable> 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,11 +173,20 @@ public void onUpdate(ClusterResource clusterResource) {
   }
 
   private void closeAllInstruments() {
+    List<RegisteredObservable> observablesToClose;
     synchronized (lock) {
-      for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
-        closeInstrument(it.next().getObservable());
-        it.remove();
-      }
+      observablesToClose = new ArrayList<>(registeredObservables);
+    }
+
+    for (RegisteredObservable registeredObservable : observablesToClose) {
+      closeInstrumentAndRemove(registeredObservable);
+    }
+  }
+
+  private void closeInstrumentAndRemove(RegisteredObservable registeredObservable) {
+    closeInstrument(registeredObservable.getObservable());
+    synchronized (lock) {
+      registeredObservables.remove(registeredObservable);
     }
   }