You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by nf...@apache.org on 2024/08/01 12:54:57 UTC
(camel) branch CAMEL-20199/remove-synchronized-blocks-from-a2c-components created (now e6d1a16fad4)
This is an automated email from the ASF dual-hosted git repository.
nfilotto pushed a change to branch CAMEL-20199/remove-synchronized-blocks-from-a2c-components
in repository https://gitbox.apache.org/repos/asf/camel.git
at e6d1a16fad4 CAMEL-20199: Remove synchronized block from components A to C
This branch includes the following new commits:
new e6d1a16fad4 CAMEL-20199: Remove synchronized block from components A to C
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-20199: Remove synchronized block from components A to C
Posted by nf...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nfilotto pushed a commit to branch CAMEL-20199/remove-synchronized-blocks-from-a2c-components
in repository https://gitbox.apache.org/repos/asf/camel.git
commit e6d1a16fad49f1feea45f1f2254edc81c1c2492e
Author: Nicolas Filotto <nf...@talend.com>
AuthorDate: Thu Aug 1 14:53:53 2024 +0200
CAMEL-20199: Remove synchronized block from components A to C
---
.../as2/api/AS2AsyncMDNServerConnection.java | 14 ++-
.../component/as2/api/AS2ServerConnection.java | 14 ++-
.../as2/internal/AS2ConnectionHelper.java | 100 +++++++++++---------
.../atmosphere/websocket/WebsocketComponent.java | 11 +--
.../component/aws2/kinesis/Kinesis2Consumer.java | 16 ++--
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 24 ++++-
.../component/bean/AbstractBeanProcessor.java | 10 +-
.../camel/component/bean/MethodInfoCache.java | 18 +---
.../component/braintree/BraintreeComponent.java | 19 ++--
.../camel/component/cometd/CometdComponent.java | 36 ++++---
.../consul/cluster/ConsulClusterView.java | 21 +++--
.../component/couchbase/CouchbaseConsumer.java | 105 +++++++++++----------
...MarshalHeaderWithCustomMarshallFactoryTest.java | 8 +-
.../camel/component/cxf/jaxrs/CxfRsProducer.java | 28 +++---
14 files changed, 234 insertions(+), 190 deletions(-)
diff --git a/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2AsyncMDNServerConnection.java b/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2AsyncMDNServerConnection.java
index 0f9660c3d95..29bc7ab8547 100644
--- a/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2AsyncMDNServerConnection.java
+++ b/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2AsyncMDNServerConnection.java
@@ -21,6 +21,8 @@ import java.io.InterruptedIOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLServerSocketFactory;
@@ -54,7 +56,7 @@ public class AS2AsyncMDNServerConnection {
private static final String REQUEST_HANDLER_THREAD_NAME_PREFIX = "AS2AsyncMdnHdlr-";
private static final int DEFAULT_BUFFER_SIZE = 8 * 1024;
private RequestListenerThread listenerThread;
- private final Object lock = new Object();
+ private final Lock lock = new ReentrantLock();
public AS2AsyncMDNServerConnection(Integer portNumber, SSLContext sslContext)
throws IOException {
@@ -66,7 +68,8 @@ public class AS2AsyncMDNServerConnection {
public void close() {
if (listenerThread != null) {
- synchronized (lock) {
+ lock.lock();
+ try {
try {
listenerThread.serverSocket.close();
} catch (IOException e) {
@@ -74,14 +77,19 @@ public class AS2AsyncMDNServerConnection {
} finally {
listenerThread = null;
}
+ } finally {
+ lock.unlock();
}
}
}
public void receive(String requestUriPattern, HttpRequestHandler handler) {
if (listenerThread != null) {
- synchronized (lock) {
+ lock.lock();
+ try {
listenerThread.registerHandler(requestUriPattern, handler);
+ } finally {
+ lock.unlock();
}
}
}
diff --git a/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ServerConnection.java b/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ServerConnection.java
index 930d0dfc05b..08976e9b122 100644
--- a/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ServerConnection.java
+++ b/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ServerConnection.java
@@ -23,6 +23,8 @@ import java.net.Socket;
import java.net.SocketException;
import java.security.PrivateKey;
import java.security.cert.Certificate;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLServerSocketFactory;
@@ -213,7 +215,7 @@ public class AS2ServerConnection {
}
private RequestListenerThread listenerThread;
- private final Object lock = new Object();
+ private final Lock lock = new ReentrantLock();
private final String as2Version;
private final String originServer;
private final String serverFqdn;
@@ -267,7 +269,8 @@ public class AS2ServerConnection {
public void close() {
if (listenerThread != null) {
- synchronized (lock) {
+ lock.lock();
+ try {
try {
listenerThread.serversocket.close();
} catch (IOException e) {
@@ -275,14 +278,19 @@ public class AS2ServerConnection {
} finally {
listenerThread = null;
}
+ } finally {
+ lock.unlock();
}
}
}
public void listen(String requestUri, HttpRequestHandler handler) {
if (listenerThread != null) {
- synchronized (lock) {
+ lock.lock();
+ try {
listenerThread.registerHandler(requestUri, handler);
+ } finally {
+ lock.unlock();
}
}
}
diff --git a/components/camel-as2/camel-as2-component/src/main/java/org/apache/camel/component/as2/internal/AS2ConnectionHelper.java b/components/camel-as2/camel-as2-component/src/main/java/org/apache/camel/component/as2/internal/AS2ConnectionHelper.java
index f19f81e6b53..fb97a485614 100644
--- a/components/camel-as2/camel-as2-component/src/main/java/org/apache/camel/component/as2/internal/AS2ConnectionHelper.java
+++ b/components/camel-as2/camel-as2-component/src/main/java/org/apache/camel/component/as2/internal/AS2ConnectionHelper.java
@@ -17,8 +17,9 @@
package org.apache.camel.component.as2.internal;
import java.io.IOException;
-import java.util.HashMap;
+import java.io.UncheckedIOException;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.camel.component.as2.AS2Configuration;
import org.apache.camel.component.as2.api.AS2AsyncMDNServerConnection;
@@ -34,9 +35,9 @@ public final class AS2ConnectionHelper {
private static final Logger LOG = LoggerFactory.getLogger(AS2ConnectionHelper.class);
- private static Map<Integer, AS2ServerConnection> serverConnections = new HashMap<>();
+ private static final Map<Integer, AS2ServerConnection> serverConnections = new ConcurrentHashMap<>();
- private static Map<Integer, AS2AsyncMDNServerConnection> asyncMdnServerConnections = new HashMap<>();
+ private static final Map<Integer, AS2AsyncMDNServerConnection> asyncMdnServerConnections = new ConcurrentHashMap<>();
/**
* Prevent instantiation
@@ -69,17 +70,20 @@ public final class AS2ConnectionHelper {
*/
public static AS2AsyncMDNServerConnection createAS2AsyncMDNServerConnection(AS2Configuration configuration)
throws IOException {
- AS2AsyncMDNServerConnection asyncMdnServerConnection
- = asyncMdnServerConnections.get(configuration.getAsyncMdnPortNumber());
- synchronized (asyncMdnServerConnections) {
- if (asyncMdnServerConnection == null) {
- asyncMdnServerConnection
- = new AS2AsyncMDNServerConnection(
- configuration.getAsyncMdnPortNumber(), configuration.getSslContext());
- asyncMdnServerConnections.put(configuration.getAsyncMdnPortNumber(), asyncMdnServerConnection);
- }
+ try {
+ return asyncMdnServerConnections.computeIfAbsent(
+ configuration.getAsyncMdnPortNumber(),
+ key -> {
+ try {
+ return new AS2AsyncMDNServerConnection(configuration.getAsyncMdnPortNumber(), configuration.getSslContext());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+ );
+ } catch (UncheckedIOException e) {
+ throw e.getCause();
}
- return asyncMdnServerConnection;
}
/**
@@ -90,18 +94,24 @@ public final class AS2ConnectionHelper {
* @throws IOException
*/
public static AS2ServerConnection createAS2ServerConnection(AS2Configuration configuration) throws IOException {
- synchronized (serverConnections) {
- AS2ServerConnection serverConnection = serverConnections.get(configuration.getServerPortNumber());
- if (serverConnection == null) {
- serverConnection = new AS2ServerConnection(
- configuration.getAs2Version(), configuration.getServer(),
- configuration.getServerFqdn(), configuration.getServerPortNumber(), configuration.getSigningAlgorithm(),
- configuration.getSigningCertificateChain(), configuration.getSigningPrivateKey(),
- configuration.getDecryptingPrivateKey(), configuration.getMdnMessageTemplate(),
- configuration.getValidateSigningCertificateChain(), configuration.getSslContext());
- serverConnections.put(configuration.getServerPortNumber(), serverConnection);
- }
- return serverConnection;
+ try {
+ return serverConnections.computeIfAbsent(
+ configuration.getServerPortNumber(),
+ key -> {
+ try {
+ return new AS2ServerConnection(
+ configuration.getAs2Version(), configuration.getServer(),
+ configuration.getServerFqdn(), configuration.getServerPortNumber(), configuration.getSigningAlgorithm(),
+ configuration.getSigningCertificateChain(), configuration.getSigningPrivateKey(),
+ configuration.getDecryptingPrivateKey(), configuration.getMdnMessageTemplate(),
+ configuration.getValidateSigningCertificateChain(), configuration.getSslContext());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+ );
+ } catch (UncheckedIOException e) {
+ throw e.getCause();
}
}
@@ -111,34 +121,30 @@ public final class AS2ConnectionHelper {
}
public static void closeAllServerConnections() {
- synchronized (serverConnections) {
- for (Map.Entry<Integer, AS2ServerConnection> entry : serverConnections.entrySet()) {
- try {
- int port = entry.getKey();
- LOG.debug("Stopping and closing AS2ServerConnection on port: {}", port);
- AS2ServerConnection conn = entry.getValue();
- conn.close();
- } catch (Exception e) {
- // ignore
- LOG.debug("Error stopping and closing AS2ServerConnection due to {}. This exception is ignored",
- e.getMessage(), e);
- }
+ for (Map.Entry<Integer, AS2ServerConnection> entry : serverConnections.entrySet()) {
+ try {
+ int port = entry.getKey();
+ LOG.debug("Stopping and closing AS2ServerConnection on port: {}", port);
+ AS2ServerConnection conn = entry.getValue();
+ conn.close();
+ } catch (Exception e) {
+ // ignore
+ LOG.debug("Error stopping and closing AS2ServerConnection due to {}. This exception is ignored",
+ e.getMessage(), e);
}
}
serverConnections.clear();
}
public static void closeAllAsyncMdnServerConnections() {
- synchronized (asyncMdnServerConnections) {
- for (Map.Entry<Integer, AS2AsyncMDNServerConnection> entry : asyncMdnServerConnections.entrySet()) {
- try {
- int port = entry.getKey();
- LOG.debug("Stopping and closing AsyncMdnServerConnection on port: {}", port);
- entry.getValue().close();
- } catch (Exception e) {
- LOG.debug("Error stopping and closing AsyncMdnServerConnection due to {}. This exception is ignored",
- e.getMessage(), e);
- }
+ for (Map.Entry<Integer, AS2AsyncMDNServerConnection> entry : asyncMdnServerConnections.entrySet()) {
+ try {
+ int port = entry.getKey();
+ LOG.debug("Stopping and closing AsyncMdnServerConnection on port: {}", port);
+ entry.getValue().close();
+ } catch (Exception e) {
+ LOG.debug("Error stopping and closing AsyncMdnServerConnection due to {}. This exception is ignored",
+ e.getMessage(), e);
}
}
asyncMdnServerConnections.clear();
diff --git a/components/camel-atmosphere-websocket/src/main/java/org/apache/camel/component/atmosphere/websocket/WebsocketComponent.java b/components/camel-atmosphere-websocket/src/main/java/org/apache/camel/component/atmosphere/websocket/WebsocketComponent.java
index 7336776a45c..137466d4810 100644
--- a/components/camel-atmosphere-websocket/src/main/java/org/apache/camel/component/atmosphere/websocket/WebsocketComponent.java
+++ b/components/camel-atmosphere-websocket/src/main/java/org/apache/camel/component/atmosphere/websocket/WebsocketComponent.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.atmosphere.websocket;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.camel.component.servlet.ServletComponent;
import org.apache.camel.component.servlet.ServletEndpoint;
@@ -29,14 +30,14 @@ import org.apache.camel.spi.annotations.Component;
*/
@Component("atmosphere-websocket")
public class WebsocketComponent extends ServletComponent {
- private Map<String, WebSocketStore> stores;
+ private final Map<String, WebSocketStore> stores;
public WebsocketComponent() {
// override the default servlet name of ServletComponent
super(WebsocketEndpoint.class);
setServletName("CamelWsServlet");
- this.stores = new HashMap<>();
+ this.stores = new ConcurrentHashMap<>();
}
@Override
@@ -46,10 +47,6 @@ public class WebsocketComponent extends ServletComponent {
}
WebSocketStore getWebSocketStore(String name) {
- WebSocketStore store;
- synchronized (stores) {
- store = stores.computeIfAbsent(name, k -> new MemoryWebSocketStore());
- }
- return store;
+ return stores.computeIfAbsent(name, k -> new MemoryWebSocketStore());
}
}
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
index 2b94b9da034..c33cedab3ba 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
@@ -16,11 +16,7 @@
*/
package org.apache.camel.component.aws2.kinesis;
-import java.util.ArrayDeque;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
+import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -56,9 +52,9 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
private KinesisConnection connection;
private ResumeStrategy resumeStrategy;
- private Map<String, String> currentShardIterators = new java.util.HashMap<>();
+ private final Map<String, String> currentShardIterators = new java.util.HashMap<>();
- private List<Shard> currentShardList = new ArrayList<>();
+ private volatile List<Shard> currentShardList = List.of();
private static final String SHARD_MONITOR_EXECUTOR_NAME = "Kinesis_shard_monitor";
private ScheduledExecutorService shardMonitorExecutor;
@@ -384,12 +380,12 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
return getEndpoint().getConfiguration();
}
- protected synchronized List<Shard> getCurrentShardList() {
+ protected List<Shard> getCurrentShardList() {
return this.currentShardList;
}
- private synchronized void setCurrentShardList(List<Shard> latestShardList) {
- this.currentShardList = latestShardList;
+ private void setCurrentShardList(List<Shard> latestShardList) {
+ this.currentShardList = List.copyOf(latestShardList);
}
private class ShardMonitor implements Runnable {
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
index 8d23db6f1a5..57c8459fb9c 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
@@ -25,6 +25,8 @@ import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
@@ -63,7 +65,7 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
private static final String TIMEOUT_CHECKER_EXECUTOR_NAME = "S3_Streaming_Upload_Timeout_Checker";
private AtomicInteger part = new AtomicInteger();
private UploadState uploadAggregate = null;
- private final Object lock = new Object();
+ private final Lock lock = new ReentrantLock();
private transient String s3ProducerToString;
private ScheduledExecutorService timeoutCheckerExecutorService;
@@ -89,11 +91,14 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
@Override
protected void doStop() throws Exception {
- synchronized (lock) {
+ lock.lock();
+ try {
if (ObjectHelper.isNotEmpty(uploadAggregate)) {
uploadPart(uploadAggregate);
completeUpload(uploadAggregate);
}
+ } finally {
+ lock.unlock();
}
if (timeoutCheckerExecutorService != null) {
getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(timeoutCheckerExecutorService);
@@ -110,12 +115,15 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
@Override
public void run() {
- synchronized (lock) {
+ lock.lock();
+ try {
if (ObjectHelper.isNotEmpty(uploadAggregate)) {
uploadPart(uploadAggregate);
completeUpload(uploadAggregate);
uploadAggregate = null;
}
+ } finally {
+ lock.unlock();
}
}
}
@@ -139,7 +147,8 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
totalSize += b.length;
if (getConfiguration().isMultiPartUpload())
maxRead -= b.length;
- synchronized (lock) {
+ lock.lock();
+ try {
// aggregate with previously received exchanges
if (ObjectHelper.isNotEmpty(uploadAggregate)) {
uploadAggregate.buffer.write(b);
@@ -165,6 +174,8 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
}
continue;
}
+ } finally {
+ lock.unlock();
}
if (state == null) {
state = new UploadState();
@@ -242,13 +253,16 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
if (ObjectHelper.isNotEmpty(state)) {
// exchange wasn't large enough to send, batch it with subsequent exchanges.
- synchronized (lock) {
+ lock.lock();
+ try {
if (ObjectHelper.isEmpty(this.uploadAggregate)) {
this.uploadAggregate = state;
} else {
// handle potential race condition.
this.uploadAggregate.buffer.write(state.buffer.toByteArray());
}
+ } finally {
+ lock.unlock();
}
}
}
diff --git a/components/camel-bean/src/main/java/org/apache/camel/component/bean/AbstractBeanProcessor.java b/components/camel-bean/src/main/java/org/apache/camel/component/bean/AbstractBeanProcessor.java
index ff287aeba53..6f184389ea5 100644
--- a/components/camel-bean/src/main/java/org/apache/camel/component/bean/AbstractBeanProcessor.java
+++ b/components/camel-bean/src/main/java/org/apache/camel/component/bean/AbstractBeanProcessor.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.bean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
import org.apache.camel.AsyncCallback;
import org.apache.camel.BeanScope;
import org.apache.camel.Exchange;
@@ -38,7 +41,7 @@ public abstract class AbstractBeanProcessor extends AsyncProcessorSupport {
private transient Processor processor;
private transient Object bean;
private transient boolean lookupProcessorDone;
- private final Object lock = new Object();
+ private final Lock lock = new ReentrantLock();
private BeanScope scope;
private String method;
private boolean shorthandMethod;
@@ -130,11 +133,14 @@ public abstract class AbstractBeanProcessor extends AsyncProcessorSupport {
boolean allowCache = scope == null || scope == BeanScope.Singleton;
if (allowCache) {
if (!lookupProcessorDone) {
- synchronized (lock) {
+ lock.lock();
+ try {
lookupProcessorDone = true;
// so if there is a custom type converter for the bean to processor
target = exchange.getContext().getTypeConverter().tryConvertTo(Processor.class, exchange, beanTmp);
processor = target;
+ } finally {
+ lock.unlock();
}
}
} else {
diff --git a/components/camel-bean/src/main/java/org/apache/camel/component/bean/MethodInfoCache.java b/components/camel-bean/src/main/java/org/apache/camel/component/bean/MethodInfoCache.java
index 896204b1121..3d52bebae7d 100644
--- a/components/camel-bean/src/main/java/org/apache/camel/component/bean/MethodInfoCache.java
+++ b/components/camel-bean/src/main/java/org/apache/camel/component/bean/MethodInfoCache.java
@@ -45,13 +45,8 @@ public class MethodInfoCache {
this.methodCache = methodCache;
}
- public synchronized MethodInfo getMethodInfo(Method method) {
- MethodInfo answer = methodCache.get(method);
- if (answer == null) {
- answer = createMethodInfo(method);
- methodCache.put(method, answer);
- }
- return answer;
+ public MethodInfo getMethodInfo(Method method) {
+ return methodCache.computeIfAbsent(method, this::createMethodInfo);
}
protected MethodInfo createMethodInfo(Method method) {
@@ -60,13 +55,8 @@ public class MethodInfoCache {
return info.getMethodInfo(method);
}
- protected synchronized BeanInfo getBeanInfo(Class<?> declaringClass) {
- BeanInfo beanInfo = classCache.get(declaringClass);
- if (beanInfo == null) {
- beanInfo = createBeanInfo(declaringClass);
- classCache.put(declaringClass, beanInfo);
- }
- return beanInfo;
+ protected BeanInfo getBeanInfo(Class<?> declaringClass) {
+ return classCache.computeIfAbsent(declaringClass, this::createBeanInfo);
}
protected BeanInfo createBeanInfo(Class<?> declaringClass) {
diff --git a/components/camel-braintree/src/main/java/org/apache/camel/component/braintree/BraintreeComponent.java b/components/camel-braintree/src/main/java/org/apache/camel/component/braintree/BraintreeComponent.java
index 509897b7700..d56b4388e1b 100644
--- a/components/camel-braintree/src/main/java/org/apache/camel/component/braintree/BraintreeComponent.java
+++ b/components/camel-braintree/src/main/java/org/apache/camel/component/braintree/BraintreeComponent.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.braintree;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import com.braintreegateway.BraintreeGateway;
import org.apache.camel.CamelContext;
@@ -41,12 +42,12 @@ public class BraintreeComponent extends AbstractApiComponent<BraintreeApiName, B
public BraintreeComponent() {
super(BraintreeApiName.class, BraintreeApiCollection.getCollection());
- this.gateways = new HashMap<>();
+ this.gateways = new ConcurrentHashMap<>();
}
public BraintreeComponent(CamelContext context) {
super(context, BraintreeApiName.class, BraintreeApiCollection.getCollection());
- this.gateways = new HashMap<>();
+ this.gateways = new ConcurrentHashMap<>();
}
@Override
@@ -63,20 +64,12 @@ public class BraintreeComponent extends AbstractApiComponent<BraintreeApiName, B
return new BraintreeEndpoint(uri, this, apiName, methodName, endpointConfiguration);
}
- public synchronized BraintreeGateway getGateway(BraintreeConfiguration configuration) {
+ public BraintreeGateway getGateway(BraintreeConfiguration configuration) {
BraintreeGateway gateway;
if (configuration.getAccessToken() != null) {
- gateway = gateways.get(configuration.getAccessToken());
- if (gateway == null) {
- gateway = configuration.newBraintreeGateway();
- gateways.put(configuration.getAccessToken(), gateway);
- }
+ gateway = gateways.computeIfAbsent(configuration.getAccessToken(), k -> configuration.newBraintreeGateway());
} else {
- gateway = gateways.get(configuration.getMerchantId());
- if (gateway == null) {
- gateway = configuration.newBraintreeGateway();
- gateways.put(configuration.getMerchantId(), gateway);
- }
+ gateway = gateways.computeIfAbsent(configuration.getMerchantId(), k -> configuration.newBraintreeGateway());
}
return gateway;
}
diff --git a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdComponent.java b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdComponent.java
index 2f912d78bc5..6aa663fd688 100644
--- a/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdComponent.java
+++ b/components/camel-cometd/src/main/java/org/apache/camel/component/cometd/CometdComponent.java
@@ -22,6 +22,8 @@ import java.util.EnumSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import jakarta.servlet.DispatcherType;
@@ -56,6 +58,7 @@ import org.slf4j.LoggerFactory;
public class CometdComponent extends DefaultComponent implements SSLContextParametersAware {
private static final Logger LOG = LoggerFactory.getLogger(CometdComponent.class);
+ private final Lock connectorsLock = new ReentrantLock();
private final Map<String, ConnectorRef> connectors = new LinkedHashMap<>();
private List<BayeuxServer.BayeuxServerListener> serverListeners;
@@ -116,7 +119,8 @@ public class CometdComponent extends DefaultComponent implements SSLContextParam
CometdEndpoint endpoint = prodcon.getEndpoint();
String connectorKey = endpoint.getProtocol() + ":" + endpoint.getUri().getHost() + ":" + endpoint.getPort();
- synchronized (connectors) {
+ connectorsLock.lock();
+ try {
ConnectorRef connectorRef = connectors.get(connectorKey);
if (connectorRef == null) {
ServerConnector connector;
@@ -160,6 +164,8 @@ public class CometdComponent extends DefaultComponent implements SSLContextParam
}
}
prodcon.setBayeux(bayeux);
+ } finally {
+ connectorsLock.unlock();
}
}
@@ -171,16 +177,17 @@ public class CometdComponent extends DefaultComponent implements SSLContextParam
String connectorKey = endpoint.getProtocol() + ":" + endpoint.getUri().getHost() + ":" + endpoint.getPort();
- synchronized (connectors) {
+ connectorsLock.lock();
+ try {
ConnectorRef connectorRef = connectors.get(connectorKey);
- if (connectorRef != null) {
- if (connectorRef.decrement() == 0) {
- connectorRef.server.removeConnector(connectorRef.connector);
- connectorRef.connector.stop();
- connectorRef.server.stop();
- connectors.remove(connectorKey);
- }
+ if (connectorRef != null && connectorRef.decrement() == 0) {
+ connectorRef.server.removeConnector(connectorRef.connector);
+ connectorRef.connector.stop();
+ connectorRef.server.stop();
+ connectors.remove(connectorKey);
}
+ } finally {
+ connectorsLock.unlock();
}
}
@@ -352,10 +359,15 @@ public class CometdComponent extends DefaultComponent implements SSLContextParam
@Override
protected void doStop() throws Exception {
- for (ConnectorRef connectorRef : connectors.values()) {
- connectorRef.connector.stop();
+ connectorsLock.lock();
+ try {
+ for (ConnectorRef connectorRef : connectors.values()) {
+ connectorRef.connector.stop();
+ }
+ connectors.clear();
+ } finally {
+ connectorsLock.unlock();
}
- connectors.clear();
super.doStop();
}
diff --git a/components/camel-consul/src/main/java/org/apache/camel/component/consul/cluster/ConsulClusterView.java b/components/camel-consul/src/main/java/org/apache/camel/component/consul/cluster/ConsulClusterView.java
index eea1972290f..d8ff7aabb6a 100644
--- a/components/camel-consul/src/main/java/org/apache/camel/component/consul/cluster/ConsulClusterView.java
+++ b/components/camel-consul/src/main/java/org/apache/camel/component/consul/cluster/ConsulClusterView.java
@@ -22,6 +22,8 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import org.apache.camel.cluster.CamelClusterMember;
@@ -44,6 +46,7 @@ final class ConsulClusterView extends AbstractCamelClusterView {
private final ConsulClusterConfiguration configuration;
private final ConsulLocalMember localMember;
+ private final Lock sessionIdLock = new ReentrantLock();
private final AtomicReference<String> sessionId;
private final Watcher watcher;
@@ -113,21 +116,24 @@ final class ConsulClusterView extends AbstractCamelClusterView {
if (keyValueClient.releaseLock(this.path, sessionId.get())) {
LOGGER.debug("Successfully released lock on path '{}' with id '{}'", path, sessionId.get());
}
-
- synchronized (sessionId) {
+ sessionIdLock.lock();
+ try {
sessionClient.destroySession(sessionId.getAndSet(null));
localMember.setMaster(false);
+ } finally {
+ sessionIdLock.unlock();
}
}
}
private boolean acquireLock() {
- synchronized (sessionId) {
+ sessionIdLock.lock();
+ try {
String sid = sessionId.get();
- return (sid != null)
- ? sessionClient.getSessionInfo(sid).map(si -> keyValueClient.acquireLock(path, sid)).orElse(Boolean.FALSE)
- : false;
+ return (sid != null) && sessionClient.getSessionInfo(sid).map(si -> keyValueClient.acquireLock(path, sid)).orElse(Boolean.FALSE);
+ } finally {
+ sessionIdLock.unlock();
}
}
@@ -136,7 +142,7 @@ final class ConsulClusterView extends AbstractCamelClusterView {
// ***********************************************
private final class ConsulLocalMember implements CamelClusterMember {
- private AtomicBoolean master = new AtomicBoolean();
+ private final AtomicBoolean master = new AtomicBoolean();
void setMaster(boolean master) {
if (master && this.master.compareAndSet(false, true)) {
@@ -147,7 +153,6 @@ final class ConsulClusterView extends AbstractCamelClusterView {
if (!master && this.master.compareAndSet(true, false)) {
LOGGER.debug("Leadership lost for session id {}", sessionId.get());
fireLeadershipChangedEvent(getLeader().orElse(null));
- return;
}
}
diff --git a/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java b/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
index 8b252244f4c..2f2ef298cc6 100644
--- a/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
+++ b/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
@@ -39,10 +39,14 @@ import static org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_ID;
import static org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_KEY;
import static org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_VIEWNAME;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
public class CouchbaseConsumer extends DefaultScheduledPollConsumer implements ResumeAware<ResumeStrategy> {
private static final Logger LOG = LoggerFactory.getLogger(CouchbaseConsumer.class);
+ private final Lock lock = new ReentrantLock();
private final CouchbaseEndpoint endpoint;
private final Bucket bucket;
private final Collection collection;
@@ -109,64 +113,69 @@ public class CouchbaseConsumer extends DefaultScheduledPollConsumer implements R
}
@Override
- protected synchronized int poll() throws Exception {
- ViewResult result = bucket.viewQuery(endpoint.getDesignDocumentName(), endpoint.getViewName(), this.viewOptions);
-
- // okay we have some response from CouchBase so lets mark the consumer as ready
- forceConsumerAsReady();
+ protected int poll() throws Exception {
+ lock.lock();
+ try {
+ ViewResult result = bucket.viewQuery(endpoint.getDesignDocumentName(), endpoint.getViewName(), this.viewOptions);
- if (LOG.isTraceEnabled()) {
- LOG.trace("ViewResponse = {}", result);
- }
+ // okay we have some response from CouchBase so lets mark the consumer as ready
+ forceConsumerAsReady();
- String consumerProcessedStrategy = endpoint.getConsumerProcessedStrategy();
- for (ViewRow row : result.rows()) {
- Object doc;
- String id = row.id().get();
- if (endpoint.isFullDocument()) {
- doc = CouchbaseCollectionOperation.getDocument(collection, id, endpoint.getQueryTimeout());
- } else {
- doc = row.valueAs(Object.class);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("ViewResponse = {}", result);
}
- String key = row.keyAs(JsonNode.class).get().asText();
- String designDocumentName = endpoint.getDesignDocumentName();
- String viewName = endpoint.getViewName();
-
- Exchange exchange = createExchange(false);
- try {
- exchange.getIn().setBody(doc);
- exchange.getIn().setHeader(HEADER_ID, id);
- exchange.getIn().setHeader(HEADER_KEY, key);
- exchange.getIn().setHeader(HEADER_DESIGN_DOCUMENT_NAME, designDocumentName);
- exchange.getIn().setHeader(HEADER_VIEWNAME, viewName);
-
- if ("delete".equalsIgnoreCase(consumerProcessedStrategy)) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Deleting doc with ID {}", id);
- }
- CouchbaseCollectionOperation.removeDocument(collection, id, endpoint.getWriteQueryTimeout(),
- endpoint.getProducerRetryPause());
- } else if ("filter".equalsIgnoreCase(consumerProcessedStrategy)) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Filtering out ID {}", id);
- }
- // add filter for already processed docs
+ String consumerProcessedStrategy = endpoint.getConsumerProcessedStrategy();
+ for (ViewRow row : result.rows()) {
+ Object doc;
+ String id = row.id().get();
+ if (endpoint.isFullDocument()) {
+ doc = CouchbaseCollectionOperation.getDocument(collection, id, endpoint.getQueryTimeout());
} else {
- LOG.trace("No strategy set for already processed docs, beware of duplicates!");
+ doc = row.valueAs(Object.class);
}
- logDetails(id, doc, key, designDocumentName, viewName, exchange);
+ String key = row.keyAs(JsonNode.class).get().asText();
+ String designDocumentName = endpoint.getDesignDocumentName();
+ String viewName = endpoint.getViewName();
+
+ Exchange exchange = createExchange(false);
+ try {
+ exchange.getIn().setBody(doc);
+ exchange.getIn().setHeader(HEADER_ID, id);
+ exchange.getIn().setHeader(HEADER_KEY, key);
+ exchange.getIn().setHeader(HEADER_DESIGN_DOCUMENT_NAME, designDocumentName);
+ exchange.getIn().setHeader(HEADER_VIEWNAME, viewName);
+
+ if ("delete".equalsIgnoreCase(consumerProcessedStrategy)) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Deleting doc with ID {}", id);
+ }
+ CouchbaseCollectionOperation.removeDocument(collection, id, endpoint.getWriteQueryTimeout(),
+ endpoint.getProducerRetryPause());
+ } else if ("filter".equalsIgnoreCase(consumerProcessedStrategy)) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Filtering out ID {}", id);
+ }
+ // add filter for already processed docs
+ } else {
+ LOG.trace("No strategy set for already processed docs, beware of duplicates!");
+ }
+
+ logDetails(id, doc, key, designDocumentName, viewName, exchange);
- getProcessor().process(exchange);
- } catch (Exception e) {
- this.getExceptionHandler().handleException("Error processing exchange.", exchange, e);
- } finally {
- releaseExchange(exchange, false);
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ this.getExceptionHandler().handleException("Error processing exchange.", exchange, e);
+ } finally {
+ releaseExchange(exchange, false);
+ }
}
- }
- return result.rows().size();
+ return result.rows().size();
+ } finally {
+ lock.unlock();
+ }
}
private void logDetails(String id, Object doc, String key, String designDocumentName, String viewName, Exchange exchange) {
diff --git a/components/camel-csv/src/test/java/org/apache/camel/dataformat/csv/CsvMarshalHeaderWithCustomMarshallFactoryTest.java b/components/camel-csv/src/test/java/org/apache/camel/dataformat/csv/CsvMarshalHeaderWithCustomMarshallFactoryTest.java
index eb16e14892d..6033d150f1d 100644
--- a/components/camel-csv/src/test/java/org/apache/camel/dataformat/csv/CsvMarshalHeaderWithCustomMarshallFactoryTest.java
+++ b/components/camel-csv/src/test/java/org/apache/camel/dataformat/csv/CsvMarshalHeaderWithCustomMarshallFactoryTest.java
@@ -27,6 +27,8 @@ import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Stream;
import org.apache.camel.Exchange;
@@ -115,6 +117,7 @@ public class CsvMarshalHeaderWithCustomMarshallFactoryTest extends CamelTestSupp
private static final class SinglePrinterCsvMarshaller extends CsvMarshaller {
+ private final Lock lock = new ReentrantLock();
private final CSVPrinter printer;
private SinglePrinterCsvMarshaller(CSVFormat format) {
@@ -134,7 +137,8 @@ public class CsvMarshalHeaderWithCustomMarshallFactoryTest extends CamelTestSupp
@Override
@SuppressWarnings("unchecked")
public void marshal(Exchange exchange, Object object, OutputStream outputStream) throws IOException {
- synchronized (printer) {
+ lock.lock();
+ try {
if (object instanceof Map) {
Map map = (Map) object;
printer.printRecord(getMapRecordValues(map));
@@ -149,6 +153,8 @@ public class CsvMarshalHeaderWithCustomMarshallFactoryTest extends CamelTestSupp
outputStream.write(stringBuilder.toString().getBytes());
// Reset the 'Appendable' for the next exchange.
stringBuilder.setLength(0);
+ } finally {
+ lock.unlock();
}
}
diff --git a/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsProducer.java b/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsProducer.java
index 809bd7c8801..de272f49175 100644
--- a/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsProducer.java
+++ b/components/camel-cxf/camel-cxf-rest/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsProducer.java
@@ -879,7 +879,7 @@ public class CxfRsProducer extends DefaultAsyncProducer {
* Cache contains {@link org.apache.cxf.jaxrs.client.JAXRSClientFactoryBean}
*/
class ClientFactoryBeanCache {
- private Map<String, JAXRSClientFactoryBean> cache;
+ private final Map<String, JAXRSClientFactoryBean> cache;
ClientFactoryBeanCache(final int maxCacheSize) {
this.cache = LRUCacheFactory.newLRUSoftCache(maxCacheSize);
@@ -896,22 +896,16 @@ public class CxfRsProducer extends DefaultAsyncProducer {
}
public JAXRSClientFactoryBean get(String address) {
- JAXRSClientFactoryBean retVal = null;
- synchronized (cache) {
- retVal = cache.get(address);
-
- if (retVal == null) {
- retVal = ((CxfRsEndpoint) getEndpoint()).createJAXRSClientFactoryBean(address);
-
- cache.put(address, retVal);
-
- LOG.trace("Created client factory bean and add to cache for address '{}'", address);
-
- } else {
- LOG.trace("Retrieved client factory bean from cache for address '{}'", address);
- }
- }
- return retVal;
+ return cache.compute(address,
+ (key, value) -> {
+ if (value == null) {
+ value = ((CxfRsEndpoint) getEndpoint()).createJAXRSClientFactoryBean(address);
+ LOG.trace("Created client factory bean and add to cache for address '{}'", address);
+ } else {
+ LOG.trace("Retrieved client factory bean from cache for address '{}'", address);
+ }
+ return value;
+ });
}
}
}