Fix concurrent modification exception in ComponentRegistry (#4951)

* Fix concurrent modification exception in ComponentRegistry * Reduce number of threads and iterations
open-telemetry · Nov 15, 2022 · bd8b565 · bd8b565
1 parent 2cca0fb
commit bd8b565
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 2 deletions.
diff --git a/sdk/common/src/main/java/io/opentelemetry/sdk/internal/ComponentRegistry.java b/sdk/common/src/main/java/io/opentelemetry/sdk/internal/ComponentRegistry.java
@@ -6,7 +6,9 @@
 package io.opentelemetry.sdk.internal;
 
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.internal.GuardedBy;
 import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.IdentityHashMap;
@@ -44,6 +46,9 @@ public final class ComponentRegistry<V> {
   private final Map<String, Map<String, Map<String, V>>> componentByNameVersionAndSchema =
       new ConcurrentHashMap<>();
 
+  private final Object lock = new Object();
+
+  @GuardedBy("lock")
   private final Set<V> allComponents = Collections.newSetFromMap(new IdentityHashMap<>());
 
   private final Function<InstrumentationScopeInfo, V> factory;
@@ -109,7 +114,9 @@ public V get(
 
   private V buildComponent(InstrumentationScopeInfo instrumentationScopeInfo) {
     V component = factory.apply(instrumentationScopeInfo);
-    allComponents.add(component);
+    synchronized (lock) {
+      allComponents.add(component);
+    }
     return component;
   }
 
@@ -119,6 +126,8 @@ private V buildComponent(InstrumentationScopeInfo instrumentationScopeInfo) {
    * @return a {@code Collection} view of the registered components.
    */
   public Collection<V> getComponents() {
-    return Collections.unmodifiableCollection(allComponents);
+    synchronized (lock) {
+      return Collections.unmodifiableCollection(new ArrayList<>(allComponents));
+    }
   }
 }
diff --git a/sdk/common/src/test/java/io/opentelemetry/sdk/internal/ComponentRegistryTest.java b/sdk/common/src/test/java/io/opentelemetry/sdk/internal/ComponentRegistryTest.java
@@ -5,9 +5,18 @@
 
 package io.opentelemetry.sdk.internal;
 
+import static java.util.stream.Collectors.joining;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import io.opentelemetry.api.common.Attributes;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.stream.IntStream;
 import org.junit.jupiter.api.Test;
 
 class ComponentRegistryTest {
@@ -52,5 +61,36 @@ void get_DifferentInstance() {
         .isNotSameAs(registry.get(NAME, null, null, Attributes.empty()));
   }
 
+  @Test
+  @SuppressWarnings("ReturnValueIgnored")
+  void getComponents_HighConcurrency() throws ExecutionException, InterruptedException {
+    List<Future<?>> futures = new ArrayList<>();
+    Random random = new Random();
+    int concurrency = 2;
+    ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+
+    try {
+      for (int i = 0; i < 100; i++) {
+        futures.add(
+            executor.submit(
+                () -> {
+                  String name =
+                      IntStream.range(0, 20)
+                          .mapToObj(unused -> String.valueOf((char) random.nextInt(26)))
+                          .collect(joining());
+                  registry.get(name, null, null, Attributes.empty());
+                }));
+        futures.add(
+            executor.submit(() -> registry.getComponents().forEach(TestComponent::hashCode)));
+      }
+
+      for (Future<?> future : futures) {
+        future.get();
+      }
+    } finally {
+      executor.shutdown();
+    }
+  }
+
   private static final class TestComponent {}
 }