diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd index 19eccf3d81209..681337b426422 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/schemas/camel-spring.xsd +++ b/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 diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractDynamicRegistry.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractDynamicRegistry.java index 7fd96abc8be71..b742fa6faca26 100644 --- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractDynamicRegistry.java +++ b/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 extends AbstractMap implements StaticService { +public class AbstractDynamicRegistry extends ConcurrentHashMap implements StaticService { protected final CamelContext context; protected final RouteController routeController; @@ -210,6 +209,11 @@ public Collection getReadOnlyValues() { return Collections.unmodifiableCollection(answer); } + @Override + public Collection values() { + return Collections.unmodifiableCollection(super.values()); + } + public Map getReadOnlyMap() { if (isEmpty()) { return Collections.emptyMap(); diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultEndpointRegistryTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultEndpointRegistryTest.java index d70fa1d996bbc..4bcae075a55f8 100644 --- a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultEndpointRegistryTest.java +++ b/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 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(); + + } + }