This is an automated email from the ASF dual-hosted git repository.
gnodet 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 a52096eea1c1 CAMEL-20199: Replace synchronized with ReentrantLock in
misc components (#25247)
a52096eea1c1 is described below
commit a52096eea1c1ece9a0ab43973232f4bdf464dce4
Author: Guillaume Nodet <[email protected]>
AuthorDate: Thu Jul 30 17:49:59 2026 +0200
CAMEL-20199: Replace synchronized with ReentrantLock in misc components
(#25247)
CAMEL-20199: Replace synchronized with ReentrantLock in misc components
Convert synchronized blocks to ReentrantLock in camel-aws2-eventbridge,
camel-azure-eventhubs, camel-event, camel-flink, camel-milo, camel-mock,
camel-netty-http, and camel-tarfile to eliminate virtual thread pinning.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../aws2/eventbridge/EventbridgeEndpoint.java | 15 +++-
.../azure/eventhubs/EventHubsConsumer.java | 94 ++++++++++++----------
.../component/camelevent/CamelEventConsumer.java | 12 ++-
.../component/flink/DataStreamFlinkProducer.java | 7 +-
.../client/MiloClientCachingConnectionManager.java | 54 ++++++++-----
.../camel/component/mock/MockExpressionClause.java | 7 +-
.../netty/http/NettyChannelBufferStreamCache.java | 11 ++-
.../tarfile/TarElementInputStreamWrapper.java | 28 ++++---
8 files changed, 144 insertions(+), 84 deletions(-)
diff --git
a/components/camel-aws/camel-aws2-eventbridge/src/main/java/org/apache/camel/component/aws2/eventbridge/EventbridgeEndpoint.java
b/components/camel-aws/camel-aws2-eventbridge/src/main/java/org/apache/camel/component/aws2/eventbridge/EventbridgeEndpoint.java
index a90ba69a69da..cafdc9b673c0 100644
---
a/components/camel-aws/camel-aws2-eventbridge/src/main/java/org/apache/camel/component/aws2/eventbridge/EventbridgeEndpoint.java
+++
b/components/camel-aws/camel-aws2-eventbridge/src/main/java/org/apache/camel/component/aws2/eventbridge/EventbridgeEndpoint.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.aws2.eventbridge;
import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.Category;
import org.apache.camel.Component;
@@ -41,6 +42,7 @@ import software.amazon.awssdk.services.sqs.SqsClient;
headersClass = EventbridgeConstants.class)
public class EventbridgeEndpoint extends ScheduledPollEndpoint implements
EndpointServiceLocation {
+ private final ReentrantLock lock = new ReentrantLock();
private EventBridgeClient eventbridgeClient;
private SqsClient sqsClient;
@@ -110,11 +112,16 @@ public class EventbridgeEndpoint extends
ScheduledPollEndpoint implements Endpoi
* Returns the SQS client used by the consumer, creating one if necessary.
Uses the same credentials and region as
* the EventBridge client.
*/
- public synchronized SqsClient getSqsClient() {
- if (sqsClient == null) {
- sqsClient = AwsClientBuilderUtil.buildClient(configuration,
SqsClient::builder);
+ public SqsClient getSqsClient() {
+ lock.lock();
+ try {
+ if (sqsClient == null) {
+ sqsClient = AwsClientBuilderUtil.buildClient(configuration,
SqsClient::builder);
+ }
+ return sqsClient;
+ } finally {
+ lock.unlock();
}
- return sqsClient;
}
@Override
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
index 21a04f735b7e..09345bdb8930 100644
---
a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
+++
b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java
@@ -22,6 +22,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
import com.azure.messaging.eventhubs.EventProcessorClient;
import com.azure.messaging.eventhubs.models.ErrorContext;
@@ -48,6 +49,7 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
// we use the EventProcessorClient as recommended by Azure docs to consume
from all partitions
private EventProcessorClient processorClient;
+ private final ReentrantLock lock = new ReentrantLock();
private final AtomicInteger pendingExchanges = new AtomicInteger();
private final Map<String, AtomicInteger> processedEventsByPartition = new
ConcurrentHashMap<>();
private final Map<String, ScheduledFuture<?>> scheduledTasksByPartition =
new ConcurrentHashMap<>();
@@ -213,52 +215,58 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
*
* @param exchange the exchange
*/
- private synchronized void processCommit(final Exchange exchange, final
EventContext eventContext) {
- final String partitionId =
eventContext.getPartitionContext().getPartitionId();
- final AtomicInteger processedEvents =
processedEventsByPartition.computeIfAbsent(partitionId,
- ignored -> new AtomicInteger());
- EventHubsCheckpointUpdaterTask checkpointTask =
checkpointTasksByPartition.get(partitionId);
- ScheduledFuture<?> scheduledTask =
scheduledTasksByPartition.get(partitionId);
-
- if (checkpointTask == null || checkpointTask.isExpired()) {
- checkpointTask = new EventHubsCheckpointUpdaterTask(eventContext,
processedEvents);
- // delegate the checkpoint update to a dedicated Thread
- long timeout = getConfiguration().getCheckpointBatchTimeout();
- checkpointTask.setScheduledTime(System.currentTimeMillis() +
timeout);
- scheduledTask = scheduledExecutorService.schedule(checkpointTask,
timeout, TimeUnit.MILLISECONDS);
- checkpointTasksByPartition.put(partitionId, checkpointTask);
- scheduledTasksByPartition.put(partitionId, scheduledTask);
- } else {
- // updates the eventContext to use for the offset to be the most
accurate
- checkpointTask.setEventContext(eventContext);
- }
-
+ private void processCommit(final Exchange exchange, final EventContext
eventContext) {
+ lock.lock();
try {
- var cnt = processedEvents.incrementAndGet();
- if (cnt == getConfiguration().getCheckpointBatchSize()) {
- processedEvents.set(0);
-
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_SIZE);
- LOG.debug("eventhub consumer batch size of reached for
partition {}", partitionId);
- if (scheduledTask != null) {
- scheduledTask.cancel(false);
- }
- eventContext.updateCheckpointAsync()
- .subscribe(unused -> LOG.debug("Processed one
event..."),
- error -> LOG.warn("Error when updating
Checkpoint: {}", error.getMessage(), error),
- () -> {
- LOG.debug("Checkpoint updated.");
- });
- } else if (checkpointTask.isExpired()) {
-
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_TIMEOUT);
- LOG.debug("eventhub consumer batch timeout reached for
partition {}", partitionId);
+ final String partitionId =
eventContext.getPartitionContext().getPartitionId();
+ final AtomicInteger processedEvents =
processedEventsByPartition.computeIfAbsent(partitionId,
+ ignored -> new AtomicInteger());
+ EventHubsCheckpointUpdaterTask checkpointTask =
checkpointTasksByPartition.get(partitionId);
+ ScheduledFuture<?> scheduledTask =
scheduledTasksByPartition.get(partitionId);
+
+ if (checkpointTask == null || checkpointTask.isExpired()) {
+ checkpointTask = new
EventHubsCheckpointUpdaterTask(eventContext, processedEvents);
+ // delegate the checkpoint update to a dedicated Thread
+ long timeout = getConfiguration().getCheckpointBatchTimeout();
+ checkpointTask.setScheduledTime(System.currentTimeMillis() +
timeout);
+ scheduledTask =
scheduledExecutorService.schedule(checkpointTask, timeout,
TimeUnit.MILLISECONDS);
+ checkpointTasksByPartition.put(partitionId, checkpointTask);
+ scheduledTasksByPartition.put(partitionId, scheduledTask);
} else {
- LOG.debug("neither eventhub consumer batch size of {}/{} nor
batch timeout reached yet for partition {}",
- cnt, getConfiguration().getCheckpointBatchSize(),
partitionId);
+ // updates the eventContext to use for the offset to be the
most accurate
+ checkpointTask.setEventContext(eventContext);
+ }
+
+ try {
+ var cnt = processedEvents.incrementAndGet();
+ if (cnt == getConfiguration().getCheckpointBatchSize()) {
+ processedEvents.set(0);
+
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_SIZE);
+ LOG.debug("eventhub consumer batch size of reached for
partition {}", partitionId);
+ if (scheduledTask != null) {
+ scheduledTask.cancel(false);
+ }
+ eventContext.updateCheckpointAsync()
+ .subscribe(unused -> LOG.debug("Processed one
event..."),
+ error -> LOG.warn("Error when updating
Checkpoint: {}", error.getMessage(), error),
+ () -> {
+ LOG.debug("Checkpoint updated.");
+ });
+ } else if (checkpointTask.isExpired()) {
+
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_TIMEOUT);
+ LOG.debug("eventhub consumer batch timeout reached for
partition {}", partitionId);
+ } else {
+ LOG.debug("neither eventhub consumer batch size of {}/{}
nor batch timeout reached yet for partition {}",
+ cnt, getConfiguration().getCheckpointBatchSize(),
partitionId);
+ }
+ // we assume that the scheduled task has done the update by
its side
+ } catch (Exception ex) {
+ getExceptionHandler().handleException(
+ "Error occurred during updating the checkpoint. This
exception is ignored.",
+ exchange, ex);
}
- // we assume that the scheduled task has done the update by its
side
- } catch (Exception ex) {
- getExceptionHandler().handleException("Error occurred during
updating the checkpoint. This exception is ignored.",
- exchange, ex);
+ } finally {
+ lock.unlock();
}
}
diff --git
a/components/camel-event/src/main/java/org/apache/camel/component/camelevent/CamelEventConsumer.java
b/components/camel-event/src/main/java/org/apache/camel/component/camelevent/CamelEventConsumer.java
index 808cdd29460a..0623eb16a773 100644
---
a/components/camel-event/src/main/java/org/apache/camel/component/camelevent/CamelEventConsumer.java
+++
b/components/camel-event/src/main/java/org/apache/camel/component/camelevent/CamelEventConsumer.java
@@ -24,6 +24,7 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
@@ -48,6 +49,7 @@ public class CamelEventConsumer extends DefaultConsumer {
private BlockingQueue<CamelEvent> eventQueue;
private ScheduledExecutorService batchScheduler;
private final List<CamelEvent> batchBuffer = new ArrayList<>();
+ private final ReentrantLock batchBufferLock = new ReentrantLock();
public CamelEventConsumer(CamelEventEndpoint endpoint, Processor
processor) {
super(endpoint, processor);
@@ -145,12 +147,15 @@ public class CamelEventConsumer extends DefaultConsumer {
*/
private void addToBatch(CamelEvent event) {
List<CamelEvent> toFlush = null;
- synchronized (batchBuffer) {
+ batchBufferLock.lock();
+ try {
batchBuffer.add(event);
if (batchBuffer.size() >= getEndpoint().getBatchSize()) {
toFlush = new ArrayList<>(batchBuffer);
batchBuffer.clear();
}
+ } finally {
+ batchBufferLock.unlock();
}
if (toFlush != null) {
processBatch(toFlush);
@@ -162,11 +167,14 @@ public class CamelEventConsumer extends DefaultConsumer {
*/
private void flushBatch() {
List<CamelEvent> toFlush = null;
- synchronized (batchBuffer) {
+ batchBufferLock.lock();
+ try {
if (!batchBuffer.isEmpty()) {
toFlush = new ArrayList<>(batchBuffer);
batchBuffer.clear();
}
+ } finally {
+ batchBufferLock.unlock();
}
if (toFlush != null) {
processBatch(toFlush);
diff --git
a/components/camel-flink/src/main/java/org/apache/camel/component/flink/DataStreamFlinkProducer.java
b/components/camel-flink/src/main/java/org/apache/camel/component/flink/DataStreamFlinkProducer.java
index 25a359694713..5ab877b20e8f 100644
---
a/components/camel-flink/src/main/java/org/apache/camel/component/flink/DataStreamFlinkProducer.java
+++
b/components/camel-flink/src/main/java/org/apache/camel/component/flink/DataStreamFlinkProducer.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.flink;
import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.Exchange;
import org.apache.camel.support.DefaultProducer;
@@ -35,6 +36,7 @@ public class DataStreamFlinkProducer extends DefaultProducer {
private static final Logger LOG =
LoggerFactory.getLogger(DataStreamFlinkProducer.class);
+ private final ReentrantLock lock = new ReentrantLock();
private volatile boolean environmentConfigured = false;
public DataStreamFlinkProducer(FlinkEndpoint endpoint) {
@@ -47,11 +49,14 @@ public class DataStreamFlinkProducer extends
DefaultProducer {
// Configure environment on first use when DataStream is available
if (!environmentConfigured && ds != null) {
- synchronized (this) {
+ lock.lock();
+ try {
if (!environmentConfigured) {
configureStreamExecutionEnvironment(ds);
environmentConfigured = true;
}
+ } finally {
+ lock.unlock();
}
}
diff --git
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientCachingConnectionManager.java
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientCachingConnectionManager.java
index e9b4f8cba722..2d8ad23434e8 100644
---
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientCachingConnectionManager.java
+++
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientCachingConnectionManager.java
@@ -20,6 +20,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
+import java.util.concurrent.locks.ReentrantLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -32,6 +33,7 @@ public class MiloClientCachingConnectionManager implements
MiloClientConnectionM
private static final Logger LOG =
LoggerFactory.getLogger(MiloClientCachingConnectionManager.class);
+ private final ReentrantLock lock = new ReentrantLock();
private final Map<String, ManagedConnection> cache = new HashMap<>();
private static class ManagedConnection {
@@ -52,33 +54,43 @@ public class MiloClientCachingConnectionManager implements
MiloClientConnectionM
}
@Override
- public synchronized MiloClientConnection createConnection(
+ public MiloClientConnection createConnection(
MiloClientConfiguration configuration,
MonitorFilterConfiguration monitorFilterConfiguration) {
- final String identifier = configuration.toCacheId();
- final ManagedConnection managedConnection
- = cache.computeIfAbsent(identifier, k ->
managedConnection(configuration, monitorFilterConfiguration));
- managedConnection.increment();
- return managedConnection.connection;
+ lock.lock();
+ try {
+ final String identifier = configuration.toCacheId();
+ final ManagedConnection managedConnection
+ = cache.computeIfAbsent(identifier, k ->
managedConnection(configuration, monitorFilterConfiguration));
+ managedConnection.increment();
+ return managedConnection.connection;
+ } finally {
+ lock.unlock();
+ }
}
@Override
- public synchronized void releaseConnection(MiloClientConnection
connection) {
- final Optional<Entry<String, ManagedConnection>> existingConnection =
this.cache.entrySet().stream()
- .filter(entry ->
entry.getValue().connection.equals(connection)).findFirst();
- existingConnection.ifPresent(entry -> {
- entry.getValue().decrement();
- if (entry.getValue().consumers <= 0) {
- try {
- LOG.debug("Closing connection {}", entry.getKey());
- entry.getValue().connection.close();
- } catch (Exception e) {
- LOG.debug("Error while closing connection with id {}. This
exception is ignored.", entry.getKey());
- } finally {
- cache.remove(entry.getKey());
+ public void releaseConnection(MiloClientConnection connection) {
+ lock.lock();
+ try {
+ final Optional<Entry<String, ManagedConnection>>
existingConnection = this.cache.entrySet().stream()
+ .filter(entry ->
entry.getValue().connection.equals(connection)).findFirst();
+ existingConnection.ifPresent(entry -> {
+ entry.getValue().decrement();
+ if (entry.getValue().consumers <= 0) {
+ try {
+ LOG.debug("Closing connection {}", entry.getKey());
+ entry.getValue().connection.close();
+ } catch (Exception e) {
+ LOG.debug("Error while closing connection with id {}.
This exception is ignored.", entry.getKey());
+ } finally {
+ cache.remove(entry.getKey());
+ }
}
- }
- });
+ });
+ } finally {
+ lock.unlock();
+ }
}
private ManagedConnection managedConnection(
diff --git
a/components/camel-mock/src/main/java/org/apache/camel/component/mock/MockExpressionClause.java
b/components/camel-mock/src/main/java/org/apache/camel/component/mock/MockExpressionClause.java
index 0138366f431e..f2a53c63d9cd 100644
---
a/components/camel-mock/src/main/java/org/apache/camel/component/mock/MockExpressionClause.java
+++
b/components/camel-mock/src/main/java/org/apache/camel/component/mock/MockExpressionClause.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.mock;
import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -38,6 +39,7 @@ import org.apache.camel.support.ExpressionToPredicateAdapter;
* are specialized for being used with the mock component and separated from
camel-core.
*/
public class MockExpressionClause<T> implements Expression, Predicate {
+ private final ReentrantLock lock = new ReentrantLock();
private final MockExpressionClauseSupport<T> delegate;
private volatile Expression expr;
@@ -429,7 +431,8 @@ public class MockExpressionClause<T> implements Expression,
Predicate {
@Override
public void init(CamelContext context) {
if (expr == null) {
- synchronized (this) {
+ lock.lock();
+ try {
if (expr == null) {
Expression newExpression = getExpressionValue();
if (newExpression == null) {
@@ -438,6 +441,8 @@ public class MockExpressionClause<T> implements Expression,
Predicate {
newExpression.init(context);
expr = newExpression;
}
+ } finally {
+ lock.unlock();
}
}
}
diff --git
a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
index 9bc782ff5a9c..1f22f51e0418 100644
---
a/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
+++
b/components/camel-netty-http/src/main/java/org/apache/camel/component/netty/http/NettyChannelBufferStreamCache.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.netty.http;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.concurrent.locks.ReentrantLock;
import io.netty.buffer.ByteBuf;
import org.apache.camel.Exchange;
@@ -36,6 +37,7 @@ import org.apache.camel.util.IOHelper;
*/
public final class NettyChannelBufferStreamCache extends InputStream
implements StreamCache {
+ private final ReentrantLock lock = new ReentrantLock();
private final ByteBuf buffer;
public NettyChannelBufferStreamCache(ByteBuf buffer) {
@@ -76,8 +78,13 @@ public final class NettyChannelBufferStreamCache extends
InputStream implements
}
@Override
- public synchronized void reset() {
- buffer.resetReaderIndex();
+ public void reset() {
+ lock.lock();
+ try {
+ buffer.resetReaderIndex();
+ } finally {
+ lock.unlock();
+ }
}
@Override
diff --git
a/components/camel-tarfile/src/main/java/org/apache/camel/dataformat/tarfile/TarElementInputStreamWrapper.java
b/components/camel-tarfile/src/main/java/org/apache/camel/dataformat/tarfile/TarElementInputStreamWrapper.java
index 84f78e39b585..604332903df7 100644
---
a/components/camel-tarfile/src/main/java/org/apache/camel/dataformat/tarfile/TarElementInputStreamWrapper.java
+++
b/components/camel-tarfile/src/main/java/org/apache/camel/dataformat/tarfile/TarElementInputStreamWrapper.java
@@ -19,6 +19,7 @@ package org.apache.camel.dataformat.tarfile;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
@@ -27,6 +28,8 @@ import
org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
*/
public class TarElementInputStreamWrapper extends BufferedInputStream {
+ private final ReentrantLock lock = new ReentrantLock();
+
public TarElementInputStreamWrapper(InputStream in, int size) {
super(in, size);
}
@@ -47,17 +50,22 @@ public class TarElementInputStreamWrapper extends
BufferedInputStream {
}
@Override
- public synchronized int available() throws IOException {
- if (in instanceof TarArchiveInputStream) {
- TarArchiveInputStream tai = (TarArchiveInputStream) in;
- if (tai.getCurrentEntry() != null) {
- // avoid NPE in TarArchiveInputStream.available which
- // only works if there is a current entry
- return tai.available();
- } else {
- return 0;
+ public int available() throws IOException {
+ lock.lock();
+ try {
+ if (in instanceof TarArchiveInputStream) {
+ TarArchiveInputStream tai = (TarArchiveInputStream) in;
+ if (tai.getCurrentEntry() != null) {
+ // avoid NPE in TarArchiveInputStream.available which
+ // only works if there is a current entry
+ return tai.available();
+ } else {
+ return 0;
+ }
}
+ return super.available();
+ } finally {
+ lock.unlock();
}
- return super.available();
}
}
