You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2023/05/19 08:31:33 UTC
[camel] branch main updated: CAMEL-19295 - Concurrency issues with dynamicMap in AbstractDynamicRegistry (#9972)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new fab07a08750 CAMEL-19295 - Concurrency issues with dynamicMap in AbstractDynamicRegistry (#9972)
fab07a08750 is described below
commit fab07a087503be96e2358d5908a40367501d5f4c
Author: Rhuan Rocha <rh...@gmail.com>
AuthorDate: Fri May 19 05:31:26 2023 -0300
CAMEL-19295 - Concurrency issues with dynamicMap in AbstractDynamicRegistry (#9972)
* CAMEL-19295 - Fixing the concurrency issues in AbstractDynamicRegistry
Signed-off-by: Rhuan Rocha <rh...@gmail.com>
* CAMEL-19295 - Fixing the concurrency issues in AbstractDynamicRegistry
Signed-off-by: Rhuan Rocha <rh...@gmail.com>
---------
Signed-off-by: Rhuan Rocha <rh...@gmail.com>
---
.../apache/camel/catalog/schemas/camel-spring.xsd | 2 +-
.../camel/impl/engine/AbstractDynamicRegistry.java | 8 +++-
.../camel/impl/DefaultEndpointRegistryTest.java | 54 ++++++++++++++++++++++
3 files changed, 61 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 4b859615e24..c4c749ab4ff 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
<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>
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 7fd96abc8be..b742fa6faca 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 @@ import org.apache.camel.support.service.ServiceHelper;
* 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 class AbstractDynamicRegistry<K, V> extends AbstractMap<K, V> implements
return Collections.unmodifiableCollection(answer);
}
+ @Override
+ public Collection<V> values() {
+ return Collections.unmodifiableCollection(super.values());
+ }
+
public Map<String, V> 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 d70fa1d996b..4bcae075a55 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 class DefaultEndpointRegistryTest {
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();
+
+ }
+
}