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:58 UTC
(camel) 01/01: CAMEL-20199: Remove synchronized block from components A to C
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;
+ });
}
}
}