You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by gn...@apache.org on 2026/07/21 13:51:22 UTC
(camel) 01/01: CAMEL-24227: Add volatile to JMX-writable fields read on routing threads
This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch fix-camel-24227-add-volatile-to-jmx-writable-fiel
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 0368046a3ff4099fbee57b6208a2e07f4ee056b1
Author: Guillaume Nodet <gn...@gmail.com>
AuthorDate: Tue Jul 21 13:51:02 2026 +0000
CAMEL-24227: Add volatile to JMX-writable fields read on routing threads
Fields writable via JMX @ManagedAttribute setters were stored as plain
(non-volatile) fields in 12 engine classes but read on routing threads
without JMM visibility guarantees.
For simple boolean/int/long fields: added volatile. On x86 this compiles
to the same instruction as a plain load, so there is zero performance
cost. On ARM/POWER architectures with weaker memory ordering, this
provides the required happens-before guarantee.
For compound writes (tracePattern+patterns, traceFilter+predicate in
BacklogTracer/DefaultTracer): replaced the two separate fields with an
immutable holder record swapped atomically via a single volatile
reference, ensuring routing threads always see a consistent pair.
Affected classes: BacklogTracer, DefaultTracer, BaseProcessorSupport,
DefaultBacklogDebugger, TotalRequestsThrottler, AbstractThrottler,
DefaultStreamCachingStrategy, ThrottlingInflightRoutePolicy,
ThrottlingExceptionRoutePolicy, ManagedPerformanceCounter,
ScheduledPollConsumer, Delayer.
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
.../apache/camel/impl/debugger/BacklogTracer.java | 77 +++++++++++++---------
.../impl/debugger/DefaultBacklogDebugger.java | 16 ++---
.../impl/engine/DefaultStreamCachingStrategy.java | 8 +--
.../apache/camel/impl/engine/DefaultTracer.java | 33 ++++++----
.../apache/camel/processor/AbstractThrottler.java | 2 +-
.../camel/processor/BaseProcessorSupport.java | 2 +-
.../java/org/apache/camel/processor/Delayer.java | 2 +-
.../camel/processor/TotalRequestsThrottler.java | 2 +-
.../mbean/ManagedPerformanceCounter.java | 2 +-
.../camel/support/ScheduledPollConsumer.java | 8 +--
.../throttling/ThrottlingExceptionRoutePolicy.java | 6 +-
.../throttling/ThrottlingInflightRoutePolicy.java | 8 +--
12 files changed, 92 insertions(+), 74 deletions(-)
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/BacklogTracer.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/BacklogTracer.java
index 059388f860d5..24b7ba04e61d 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/BacklogTracer.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/BacklogTracer.java
@@ -62,13 +62,13 @@ public class BacklogTracer extends ServiceSupport implements org.apache.camel.sp
public static final int MAX_BACKLOG_SIZE = 1000;
private final CamelContext camelContext;
private final Language simple;
- private boolean enabled;
- private boolean standby;
+ private volatile boolean enabled;
+ private volatile boolean standby;
private final AtomicLong traceCounter = new AtomicLong();
// use a queue with an upper limit to avoid storing too many messages
private final Queue<BacklogTracerEventMessage> queue = new LinkedBlockingQueue<>(MAX_BACKLOG_SIZE);
// how many of the last messages to keep in the backlog at total
- private int backlogSize = 100;
+ private volatile int backlogSize = 100;
// use tracer to capture additional information for capturing latest completed exchange message-history
private final Queue<BacklogTracerEventMessage> provisionalHistoryQueue = new LinkedBlockingQueue<>(MAX_BACKLOG_SIZE);
private final Queue<BacklogTracerEventMessage> completeHistoryQueue = new LinkedBlockingQueue<>(MAX_BACKLOG_SIZE + 1);
@@ -76,25 +76,31 @@ public class BacklogTracer extends ServiceSupport implements org.apache.camel.sp
private final Queue<BacklogTracerActivityMessage> activityQueue = new LinkedBlockingQueue<>(MAX_BACKLOG_SIZE);
private final ConcurrentHashMap<String, DefaultBacklogTracerActivityMessage> inflightActivity = new ConcurrentHashMap<>();
private final ActivityEventNotifier activityEventNotifier = new ActivityEventNotifier();
- private int activitySize = 100;
+ private volatile int activitySize = 100;
private static final long INFLIGHT_EVICTION_MILLIS = 5 * 60 * 1000;
private final Object historyLock = new Object();
private volatile String lastCompletedBreadcrumbId;
- private boolean removeOnDump = true;
- private int bodyMaxChars = 32 * 1024;
- private boolean bodyIncludeStreams;
- private boolean bodyIncludeFiles = true;
- private boolean includeExchangeProperties = true;
- private boolean includeExchangeVariables = true;
- private boolean includeException = true;
- private boolean activityEnabled;
- private boolean traceRests;
- private boolean traceTemplates;
+ private volatile boolean removeOnDump = true;
+ private volatile int bodyMaxChars = 32 * 1024;
+ private volatile boolean bodyIncludeStreams;
+ private volatile boolean bodyIncludeFiles = true;
+ private volatile boolean includeExchangeProperties = true;
+ private volatile boolean includeExchangeVariables = true;
+ private volatile boolean includeException = true;
+ private volatile boolean activityEnabled;
+ private volatile boolean traceRests;
+ private volatile boolean traceTemplates;
+
+ // immutable holders for compound fields that must be visible atomically on routing threads
+ private record TracePatternHolder(String tracePattern, String[] patterns) {
+ }
+
+ private record TraceFilterHolder(String traceFilter, Predicate predicate) {
+ }
+
// a pattern to filter tracing nodes
- private String tracePattern;
- private String[] patterns;
- private String traceFilter;
- private Predicate predicate;
+ private volatile TracePatternHolder tracePatternHolder;
+ private volatile TraceFilterHolder traceFilterHolder;
BacklogTracer(CamelContext camelContext) {
this.camelContext = camelContext;
@@ -130,17 +136,20 @@ public class BacklogTracer extends ServiceSupport implements org.apache.camel.sp
boolean pattern = true;
boolean filter = true;
- if (patterns != null) {
- pattern = shouldTracePattern(definition);
+ // snapshot the volatile holders once for consistent reads
+ TracePatternHolder ph = tracePatternHolder;
+ if (ph != null) {
+ pattern = shouldTracePattern(definition, ph.patterns());
}
- if (predicate != null) {
- filter = shouldTraceFilter(exchange);
+ TraceFilterHolder fh = traceFilterHolder;
+ if (fh != null) {
+ filter = shouldTraceFilter(exchange, fh.predicate());
}
return pattern && filter;
}
- private boolean shouldTracePattern(NamedNode definition) {
+ private boolean shouldTracePattern(NamedNode definition, String[] patterns) {
for (String pattern : patterns) {
// match either route id, or node id
String id = definition.getId();
@@ -321,7 +330,7 @@ public class BacklogTracer extends ServiceSupport implements org.apache.camel.sp
}
}
- private boolean shouldTraceFilter(Exchange exchange) {
+ private boolean shouldTraceFilter(Exchange exchange, Predicate predicate) {
return predicate.matches(exchange);
}
@@ -473,37 +482,41 @@ public class BacklogTracer extends ServiceSupport implements org.apache.camel.sp
@Override
public String getTracePattern() {
- return tracePattern;
+ TracePatternHolder ph = tracePatternHolder;
+ return ph != null ? ph.tracePattern() : null;
}
@Override
public void setTracePattern(String tracePattern) {
- this.tracePattern = tracePattern;
if (tracePattern != null) {
// the pattern can have multiple nodes separated by comma
- this.patterns = tracePattern.split(",");
+ this.tracePatternHolder = new TracePatternHolder(tracePattern, tracePattern.split(","));
} else {
- this.patterns = null;
+ this.tracePatternHolder = null;
}
}
@Override
public String getTraceFilter() {
- return traceFilter;
+ TraceFilterHolder fh = traceFilterHolder;
+ return fh != null ? fh.traceFilter() : null;
}
@Override
public void setTraceFilter(String filter) {
- this.traceFilter = filter;
if (filter != null) {
// assume simple language
+ Predicate p;
String name = StringHelper.before(filter, ":");
if (name != null) {
- predicate = camelContext.resolveLanguage(name).createPredicate(filter);
+ p = camelContext.resolveLanguage(name).createPredicate(filter);
} else {
// use simple language by default
- predicate = simple.createPredicate(filter);
+ p = simple.createPredicate(filter);
}
+ this.traceFilterHolder = new TraceFilterHolder(filter, p);
+ } else {
+ this.traceFilterHolder = null;
}
}
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogDebugger.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogDebugger.java
index 09d9aad0e903..9eed5c9af64e 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogDebugger.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/debugger/DefaultBacklogDebugger.java
@@ -68,7 +68,7 @@ public final class DefaultBacklogDebugger extends ServiceSupport implements Back
private static final Logger LOG = LoggerFactory.getLogger(DefaultBacklogDebugger.class);
- private long fallbackTimeout = 300;
+ private volatile long fallbackTimeout = 300;
private final CamelContext camelContext;
private LoggingLevel loggingLevel = LoggingLevel.INFO;
private final CamelLogger logger = new CamelLogger(LOG, loggingLevel);
@@ -87,13 +87,13 @@ public final class DefaultBacklogDebugger extends ServiceSupport implements Back
private boolean suspendMode;
private String initialBreakpoints;
- private boolean singleStepIncludeStartEnd;
- private int bodyMaxChars = 32 * 1024;
- private boolean bodyIncludeStreams;
- private boolean bodyIncludeFiles = true;
- private boolean includeExchangeProperties = true;
- private boolean includeExchangeVariables = true;
- private boolean includeException = true;
+ private volatile boolean singleStepIncludeStartEnd;
+ private volatile int bodyMaxChars = 32 * 1024;
+ private volatile boolean bodyIncludeStreams;
+ private volatile boolean bodyIncludeFiles = true;
+ private volatile boolean includeExchangeProperties = true;
+ private volatile boolean includeExchangeVariables = true;
+ private volatile boolean includeException = true;
/**
* An {@link Exchange} suspended at a breakpoint.
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultStreamCachingStrategy.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultStreamCachingStrategy.java
index ddc41bf45d44..377ff8fef942 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultStreamCachingStrategy.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultStreamCachingStrategy.java
@@ -64,15 +64,15 @@ public class DefaultStreamCachingStrategy extends ServiceSupport implements Came
private boolean spoolEnabled;
private File spoolDirectory;
private transient String spoolDirectoryName = "${java.io.tmpdir}/camel/camel-tmp-#uuid#";
- private long spoolThreshold = StreamCache.DEFAULT_SPOOL_THRESHOLD;
- private int spoolUsedHeapMemoryThreshold;
+ private volatile long spoolThreshold = StreamCache.DEFAULT_SPOOL_THRESHOLD;
+ private volatile int spoolUsedHeapMemoryThreshold;
private SpoolUsedHeapMemoryLimit spoolUsedHeapMemoryLimit;
private String spoolCipher;
- private int bufferSize = IOHelper.DEFAULT_BUFFER_SIZE;
+ private volatile int bufferSize = IOHelper.DEFAULT_BUFFER_SIZE;
private boolean removeSpoolDirectoryWhenStopping = true;
private final UtilizationStatistics statistics = new UtilizationStatistics();
private final Set<SpoolRule> spoolRules = new LinkedHashSet<>();
- private boolean anySpoolRules;
+ private volatile boolean anySpoolRules;
@Override
public CamelContext getCamelContext() {
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultTracer.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultTracer.java
index 3a76176ca21a..f01d9af6b51a 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultTracer.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultTracer.java
@@ -55,16 +55,19 @@ public class DefaultTracer extends ServiceSupport implements CamelContextAware,
private String tracingFormat = TRACING_OUTPUT;
private CamelContext camelContext;
- private boolean enabled = true;
- private boolean standby;
- private boolean traceRests;
- private boolean traceTemplates;
+ private volatile boolean enabled = true;
+ private volatile boolean standby;
+ private volatile boolean traceRests;
+ private volatile boolean traceTemplates;
private long traceCounter;
+ // immutable holder for compound tracePattern+patterns that must be visible atomically
+ private record TracePatternHolder(String tracePattern, String[] patterns) {
+ }
+
private ExchangeFormatter exchangeFormatter;
- private String tracePattern;
- private transient String[] patterns;
- private boolean traceBeforeAndAfterRoute = true;
+ private volatile TracePatternHolder tracePatternHolder;
+ private volatile boolean traceBeforeAndAfterRoute = true;
public DefaultTracer() {
DefaultExchangeFormatter formatter = new DefaultExchangeFormatter();
@@ -241,8 +244,10 @@ public class DefaultTracer extends ServiceSupport implements CamelContextAware,
boolean pattern = true;
- if (patterns != null) {
- pattern = shouldTracePattern(definition);
+ // snapshot the volatile holder once for consistent reads
+ TracePatternHolder ph = tracePatternHolder;
+ if (ph != null) {
+ pattern = shouldTracePattern(definition, ph.patterns());
}
if (LOG.isTraceEnabled()) {
@@ -303,17 +308,17 @@ public class DefaultTracer extends ServiceSupport implements CamelContextAware,
@Override
public String getTracePattern() {
- return tracePattern;
+ TracePatternHolder ph = tracePatternHolder;
+ return ph != null ? ph.tracePattern() : null;
}
@Override
public void setTracePattern(String tracePattern) {
- this.tracePattern = tracePattern;
if (tracePattern != null) {
// the pattern can have multiple nodes separated by comma
- this.patterns = tracePattern.split(",");
+ this.tracePatternHolder = new TracePatternHolder(tracePattern, tracePattern.split(","));
} else {
- this.patterns = null;
+ this.tracePatternHolder = null;
}
}
@@ -347,7 +352,7 @@ public class DefaultTracer extends ServiceSupport implements CamelContextAware,
}
}
- protected boolean shouldTracePattern(NamedNode definition) {
+ protected boolean shouldTracePattern(NamedNode definition, String[] patterns) {
for (String pattern : patterns) {
// match either route id, or node id
String id = definition.getId();
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/AbstractThrottler.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/AbstractThrottler.java
index 947dfdf990b7..725e51122aa8 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/AbstractThrottler.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/AbstractThrottler.java
@@ -49,7 +49,7 @@ public abstract class AbstractThrottler extends BaseProcessorSupport
protected boolean rejectExecution;
protected boolean asyncDelayed;
protected boolean callerRunsWhenRejected = true;
- protected Expression maxRequestsExpression;
+ protected volatile Expression maxRequestsExpression;
AbstractThrottler(final ScheduledExecutorService asyncExecutor, final boolean shutdownAsyncExecutor,
final CamelContext camelContext, final boolean rejectExecution, Expression correlation,
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/BaseProcessorSupport.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/BaseProcessorSupport.java
index ed2b416ce1e6..1329b0a18fd9 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/BaseProcessorSupport.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/BaseProcessorSupport.java
@@ -24,7 +24,7 @@ import org.apache.camel.support.AsyncProcessorSupport;
*/
public abstract class BaseProcessorSupport extends AsyncProcessorSupport implements DisabledAware {
- private boolean disabled;
+ private volatile boolean disabled;
@Override
public boolean isDisabled() {
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Delayer.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Delayer.java
index 9e588400e9a2..98043650072b 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Delayer.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Delayer.java
@@ -38,7 +38,7 @@ public class Delayer extends DelayProcessorSupport implements Traceable, IdAware
private String routeId;
private String stepId;
private String id;
- private Expression delay;
+ private volatile Expression delay;
private long delayValue;
public Delayer(CamelContext camelContext, Processor processor, Expression delay,
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
index aa89376ed5a0..2cd000321e1a 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
@@ -58,7 +58,7 @@ public class TotalRequestsThrottler extends AbstractThrottler {
private static final Logger LOG = LoggerFactory.getLogger(TotalRequestsThrottler.class);
- private long timePeriodMillis;
+ private volatile long timePeriodMillis;
private final long cleanPeriodMillis;
private final Expression correlationExpression;
private final Map<String, ThrottlingState> states = new ConcurrentHashMap<>();
diff --git a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
index 6b7187eb27ef..9bca27e455b6 100644
--- a/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
+++ b/core/camel-management/src/main/java/org/apache/camel/management/mbean/ManagedPerformanceCounter.java
@@ -61,7 +61,7 @@ public abstract class ManagedPerformanceCounter extends ManagedCounter
private Statistic lastExchangeFailureTimestamp;
private String lastExchangeFailureExchangeId;
private final LoadThroughput thp = new LoadThroughput();
- private boolean statisticsEnabled = true;
+ private volatile boolean statisticsEnabled = true;
// sliding window ring buffer for percentile computation (Extended statistics only)
private long[] percentileWindow;
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java b/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
index 016c1994c98e..878180552ad5 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
@@ -57,13 +57,13 @@ public abstract class ScheduledPollConsumer extends DefaultConsumer
private boolean startScheduler = true;
private long initialDelay = 1000;
- private long delay = 500;
+ private volatile long delay = 500;
private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
private boolean useFixedDelay = true;
private PollingConsumerPollStrategy pollStrategy;
- private LoggingLevel runLoggingLevel = LoggingLevel.TRACE;
- private boolean sendEmptyMessageWhenIdle;
- private boolean greedy;
+ private volatile LoggingLevel runLoggingLevel = LoggingLevel.TRACE;
+ private volatile boolean sendEmptyMessageWhenIdle;
+ private volatile boolean greedy;
private int backoffMultiplier;
private int backoffIdleThreshold;
private int backoffErrorThreshold;
diff --git a/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java b/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
index 0595eed5bc43..1254990c99fe 100644
--- a/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
+++ b/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
@@ -73,13 +73,13 @@ public class ThrottlingExceptionRoutePolicy extends RoutePolicySupport implement
// configuration
@Metadata(description = "How many failed messages within the window would trigger the circuit breaker to open",
defaultValue = "50")
- private int failureThreshold = 50;
+ private volatile int failureThreshold = 50;
@Metadata(description = "Sliding window for how long time to go back (in millis) when counting number of failures",
defaultValue = "60000")
- private long failureWindow = 60000;
+ private volatile long failureWindow = 60000;
@Metadata(description = "Interval (in millis) for how often to check whether a currently open circuit breaker may work again",
defaultValue = "30000")
- private long halfOpenAfter = 30000;
+ private volatile long halfOpenAfter = 30000;
@Metadata(description = "Whether to always keep the circuit breaker open (never closes). This is only intended for development and testing purposes.")
private boolean keepOpen;
@Metadata(description = "Allows to only throttle based on certain types of exceptions. Multiple exceptions (use FQN class name) can be separated by comma.")
diff --git a/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingInflightRoutePolicy.java b/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingInflightRoutePolicy.java
index 9cb030b919c6..8adf9bc0e434 100644
--- a/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingInflightRoutePolicy.java
+++ b/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingInflightRoutePolicy.java
@@ -71,14 +71,14 @@ public class ThrottlingInflightRoutePolicy extends RoutePolicySupport implements
private final Lock lock = new ReentrantLock();
@Metadata(description = "Sets which scope the throttling should be based upon, either route or total scoped.",
enums = "Context,Route", defaultValue = "Route")
- private ThrottlingScope scope = ThrottlingScope.Route;
+ private volatile ThrottlingScope scope = ThrottlingScope.Route;
@Metadata(description = "Sets the upper limit of number of concurrent inflight exchanges at which point reached the throttler should suspend the route.",
defaultValue = "1000")
- private int maxInflightExchanges = 1000;
+ private volatile int maxInflightExchanges = 1000;
@Metadata(description = "Sets at which percentage of the max the throttler should start resuming the route.",
defaultValue = "70")
- private int resumePercentOfMax = 70;
- private int resumeInflightExchanges = 700;
+ private volatile int resumePercentOfMax = 70;
+ private volatile int resumeInflightExchanges = 700;
@Metadata(description = "Sets the logging level to report the throttling activity.",
javaType = "org.apache.camel.LoggingLevel", defaultValue = "INFO", enums = "TRACE,DEBUG,INFO,WARN,ERROR,OFF")
private LoggingLevel loggingLevel = LoggingLevel.INFO;