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 <[email protected]> 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 <[email protected]> --- .../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;
