From f7f7aebacc6edd5ceb9b530289ac4d17fc0dd126 Mon Sep 17 00:00:00 2001 From: Rhuan Rocha Date: Mon, 1 May 2023 19:58:52 -0300 Subject: [PATCH 1/2] CAMEL-19295 - Fixing the concurrency issues in AbstractDynamicRegistry Signed-off-by: Rhuan Rocha --- .../camel/catalog/schemas/camel-spring.xsd | 2 +- .../impl/engine/AbstractDynamicRegistry.java | 8 ++- .../impl/DefaultEndpointRegistryTest.java | 55 +++++++++++++++++++ 3 files changed, 62 insertions(+), 3 deletions(-) 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..7bd8c4311ed2d 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,11 +16,20 @@ */ 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; public class DefaultEndpointRegistryTest { @@ -68,4 +77,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(); + + } + } From 74feead19959840689187abc769816115eb68848 Mon Sep 17 00:00:00 2001 From: Rhuan Rocha Date: Mon, 1 May 2023 20:25:01 -0300 Subject: [PATCH 2/2] CAMEL-19295 - Fixing the concurrency issues in AbstractDynamicRegistry Signed-off-by: Rhuan Rocha --- .../org/apache/camel/impl/DefaultEndpointRegistryTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 7bd8c4311ed2d..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 @@ -29,7 +29,6 @@ import org.apache.camel.support.NormalizedUri; import org.junit.jupiter.api.Test; - import static org.junit.jupiter.api.Assertions.assertTrue; public class DefaultEndpointRegistryTest { @@ -115,7 +114,7 @@ public void testConcurrency() throws InterruptedException { allThreadCompletionSemaphore.await(); - assertTrue( endpointRegistry.values().toArray() != null); + assertTrue(endpointRegistry.values().toArray() != null); }