CAMEL-19295 - Concurrency issues with dynamicMap in AbstractDynamicRegistry

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd

@@ -384,7 +384,7 @@ Enriches a message with data from a secondary resource
   <xs:element name="errorHandler" nillable="true" type="xs:anyType">
     <xs:annotation>
       <xs:documentation xml:lang="en"><![CDATA[
-Camel error handling.
+Error handler settings
       ]]></xs:documentation>
     </xs:annotation>
   </xs:element>

core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractDynamicRegistry.java

@@ -16,7 +16,6 @@
  */
 package org.apache.camel.impl.engine;
 
-import java.util.AbstractMap;
 import java.util.AbstractSet;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -40,7 +39,7 @@
  * Base implementation for {@link org.apache.camel.spi.EndpointRegistry},
  * {@link org.apache.camel.spi.TransformerRegistry}, and {@link org.apache.camel.spi.ValidatorRegistry}.
  */
-public class AbstractDynamicRegistry<K, V> extends AbstractMap<K, V> implements StaticService {
+public class AbstractDynamicRegistry<K, V> extends ConcurrentHashMap<K, V> implements StaticService {
 
     protected final CamelContext context;
     protected final RouteController routeController;
@@ -210,6 +209,11 @@ public Collection<V> getReadOnlyValues() {
         return Collections.unmodifiableCollection(answer);
     }
 
+    @Override
+    public Collection<V> values() {
+        return Collections.unmodifiableCollection(super.values());
+    }
+
     public Map<String, V> getReadOnlyMap() {
         if (isEmpty()) {
             return Collections.emptyMap();

core/camel-core/src/test/java/org/apache/camel/impl/DefaultEndpointRegistryTest.java

@@ -16,9 +16,17 @@
  */
 package org.apache.camel.impl;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.ServiceStatus;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.impl.engine.DefaultEndpointRegistry;
+import org.apache.camel.impl.engine.SimpleCamelContext;
 import org.apache.camel.spi.EndpointRegistry;
+import org.apache.camel.support.NormalizedUri;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -68,4 +76,50 @@ public void configure() throws Exception {
         assertTrue(reg.isStatic("file:error"));
     }
 
+    //Testing the issue https://issues.apache.org/jira/browse/CAMEL-19295
+    @Test
+    public void testConcurrency() throws InterruptedException {
+
+        SimpleCamelContext context = new SimpleCamelContext();
+        context.start();
+
+        ProducerTemplate producerTemplate = context.createProducerTemplate();
+        EndpointRegistry<NormalizedUri> endpointRegistry = context.getEndpointRegistry();
+
+        int nThreads = 4;
+        ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
+        int iterations = 500;
+
+        for (int j = 0; j < iterations; j++) {
+            CountDownLatch allThreadCompletionSemaphore = new CountDownLatch(nThreads);
+            for (int i = 0; i < nThreads; i++) {
+
+                executorService.submit(() -> {
+
+                    producerTemplate.requestBody("controlbus:route?routeId=route1&action=ACTION_STATUS&loggingLevel=off", null,
+                            ServiceStatus.class);
+                    producerTemplate.requestBody("controlbus:route?routeId=route2&action=ACTION_STATUS&loggingLevel=off", null,
+                            ServiceStatus.class);
+                    producerTemplate.requestBody("controlbus:route?routeId=route3&action=ACTION_STATUS&loggingLevel=off", null,
+                            ServiceStatus.class);
+                    producerTemplate.requestBody("controlbus:route?routeId=route4&action=ACTION_STATUS&loggingLevel=off", null,
+                            ServiceStatus.class);
+                    producerTemplate.requestBody("controlbus:route?routeId=route5&action=ACTION_STATUS&loggingLevel=off", null,
+                            ServiceStatus.class);
+
+                    allThreadCompletionSemaphore.countDown();
+
+                });
+            }
+
+            allThreadCompletionSemaphore.await();
+
+            assertTrue(endpointRegistry.values().toArray() != null);
+
+        }
+
+        executorService.shutdown();
+
+    }
+
 }