fix: kafka initialization occasionally failed due to concurrent injec…

…tion of OpenTelemetryMetricsReporter. (#12538)
open-telemetry · Nov 7, 2024 · 719e4ee · 719e4ee
1 parent c8bd230
commit 719e4ee
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 8 deletions.
diff --git a/...agent/instrumentation/kafkaclients/v0_11/metrics/KafkaMetricsConsumerInstrumentation.java b/...agent/instrumentation/kafkaclients/v0_11/metrics/KafkaMetricsConsumerInstrumentation.java
@@ -42,10 +42,24 @@ public static class ConstructorMapAdvice {
     @Advice.OnMethodEnter(suppress = Throwable.class)
     public static void onEnter(
         @Advice.Argument(value = 0, readOnly = false) Map<String, Object> config) {
-      // ensure config is a mutable map
-      if (config.getClass() != HashMap.class) {
-        config = new HashMap<>(config);
-      }
+
+      // In versions of spring-kafka prior to 2.5.0.RC1, when the `ProducerPerThread`
+      //  of DefaultKafkaProducerFactory is set to true, the `config` object entering
+      //  this advice block can be shared across multiple threads. Directly modifying
+      //  `config` could lead to unexpected item loss due to race conditions, where
+      //  some entries might be lost as different threads attempt to modify it
+      //  concurrently.
+      //
+      // To prevent such issues, a copy of the `config` should be created here before
+      //  any modifications are made. This ensures that each thread operates on its
+      //  own independent copy of the configuration, thereby eliminating the risk of
+      //  configurations corruption.
+      //
+      // More detailed information:
+      //  https://github.com/open-telemetry/opentelemetry-java-instrumentation/issues/12538
+
+      // ensure config is a mutable map and avoid concurrency conflicts
+      config = new HashMap<>(config);
       enhanceConfig(config);
     }
   }

diff --git a/...agent/instrumentation/kafkaclients/v0_11/metrics/KafkaMetricsProducerInstrumentation.java b/...agent/instrumentation/kafkaclients/v0_11/metrics/KafkaMetricsProducerInstrumentation.java
@@ -42,10 +42,24 @@ public static class ConstructorMapAdvice {
     @Advice.OnMethodEnter(suppress = Throwable.class)
     public static void onEnter(
         @Advice.Argument(value = 0, readOnly = false) Map<String, Object> config) {
-      // ensure config is a mutable map
-      if (config.getClass() != HashMap.class) {
-        config = new HashMap<>(config);
-      }
+
+      // In versions of spring-kafka prior to 2.5.0.RC1, when the `ProducerPerThread`
+      //  of DefaultKafkaProducerFactory is set to true, the `config` object entering
+      //  this advice block can be shared across multiple threads. Directly modifying
+      //  `config` could lead to unexpected item loss due to race conditions, where
+      //  some entries might be lost as different threads attempt to modify it
+      //  concurrently.
+      //
+      // To prevent such issues, a copy of the `config` should be created here before
+      //  any modifications are made. This ensures that each thread operates on its
+      //  own independent copy of the configuration, thereby eliminating the risk of
+      //  configurations corruption.
+      //
+      // More detailed information:
+      //  https://github.com/open-telemetry/opentelemetry-java-instrumentation/issues/12538
+
+      // ensure config is a mutable map and avoid concurrency conflicts
+      config = new HashMap<>(config);
       enhanceConfig(config);
     }
   }