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 23b6f520ff32 CAMEL-24227: Add volatile to JMX-writable fields read on
routing threads
23b6f520ff32 is described below
commit 23b6f520ff32eb41c2da70aebeaabf044dc9411b
Author: Guillaume Nodet <[email protected]>
AuthorDate: Mon Jul 27 15:20:24 2026 +0200
CAMEL-24227: Add volatile to JMX-writable fields read on routing threads
Fix JMM-unsafe patterns across 15 engine classes where fields written
at runtime via JMX/management APIs lacked memory visibility guarantees
for routing threads. Simple fields get volatile; compound writes
(BacklogTracer trace pattern/filter, DefaultTracer trace pattern,
ThrottlingInflightRoutePolicy limits) use immutable holder records
swapped via a single volatile reference. Also fixes a pre-existing bug
where setTraceFilter(null) left a stale predicate in BacklogTracer.
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../apache/camel/impl/debugger/BacklogTracer.java | 78 ++++++++------
.../impl/debugger/DefaultBacklogDebugger.java | 16 +--
.../impl/engine/DefaultStreamCachingStrategy.java | 10 +-
.../apache/camel/impl/engine/DefaultTracer.java | 33 +++---
.../apache/camel/processor/AbstractThrottler.java | 2 +-
.../processor/BaseDelegateProcessorSupport.java | 2 +-
.../camel/processor/BaseProcessorSupport.java | 2 +-
.../java/org/apache/camel/processor/Delayer.java | 2 +-
.../camel/processor/TotalRequestsThrottler.java | 2 +-
.../org/apache/camel/processor/WrapProcessor.java | 2 +-
.../mbean/ManagedPerformanceCounter.java | 2 +-
.../management/BacklogTracerFilterClearTest.java | 116 +++++++++++++++++++++
.../camel/support/ScheduledPollConsumer.java | 14 +--
.../throttling/ThrottlingExceptionRoutePolicy.java | 6 +-
.../throttling/ThrottlingInflightRoutePolicy.java | 50 ++++++---
15 files changed, 243 insertions(+), 94 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 90fc97af0823..8399000cc572 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,17 +62,15 @@ 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;
- // enabled, standby, and activityEnabled (further below) are toggled at
runtime via
- // JMX/management APIs while routing threads read them in shouldTrace()
and traceEvent().
- // Other boolean fields (removeOnDump, bodyIncludeStreams, traceRests,
etc.) are set during
- // initialization and do not change while routes are processing, so they
do not need volatile.
+ // Fields below marked volatile are writable at runtime via JMX/management
APIs
+ // while routing threads read them in shouldTrace() and traceEvent().
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);
@@ -80,26 +78,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;
- // volatile: toggled at runtime via JMX, same rationale as enabled/standby
above
+ 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 boolean traceRests;
- private boolean traceTemplates;
+ 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;
@@ -135,17 +138,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();
@@ -326,7 +332,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);
}
@@ -478,37 +484,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 f64aec981ac7..b0c93e6bbd2b 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() {
@@ -561,7 +561,7 @@ public class DefaultStreamCachingStrategy extends
ServiceSupport implements Came
private static final class UtilizationStatistics implements Statistics {
private final Lock lock = new ReentrantLock();
- private boolean statisticsEnabled;
+ private volatile boolean statisticsEnabled;
private final AtomicLong memoryCounter = new AtomicLong();
private final AtomicLong memorySize = new AtomicLong();
private final AtomicLong memoryAverageSize = new AtomicLong();
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/BaseDelegateProcessorSupport.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/BaseDelegateProcessorSupport.java
index 52a5229fb72c..7a77d6b373ff 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/BaseDelegateProcessorSupport.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/BaseDelegateProcessorSupport.java
@@ -28,7 +28,7 @@ public abstract class BaseDelegateProcessorSupport extends
DelegateAsyncProcesso
super(processor);
}
- 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/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-core-processor/src/main/java/org/apache/camel/processor/WrapProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/WrapProcessor.java
index 703eb8664be7..da510d1033dc 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/WrapProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/WrapProcessor.java
@@ -29,7 +29,7 @@ import org.apache.camel.support.service.ServiceHelper;
public class WrapProcessor extends DelegateAsyncProcessor implements
WrapAwareProcessor {
private final Processor wrapped;
- private boolean disabled;
+ private volatile boolean disabled;
public WrapProcessor(Processor processor, Processor wrapped) {
super(processor);
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-management/src/test/java/org/apache/camel/management/BacklogTracerFilterClearTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/BacklogTracerFilterClearTest.java
new file mode 100644
index 000000000000..1192d4ac8580
--- /dev/null
+++
b/core/camel-management/src/test/java/org/apache/camel/management/BacklogTracerFilterClearTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.management;
+
+import java.util.List;
+
+import javax.management.Attribute;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.BacklogTracerEventMessage;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@DisabledOnOs(OS.AIX)
+class BacklogTracerFilterClearTest extends ManagementTestSupport {
+
+ /**
+ * Verify that clearing the trace filter via JMX (setTraceFilter(null))
actually disables filtering so that
+ * subsequent messages are traced unconditionally. Before the fix, the old
code nulled the filter string but never
+ * cleared the predicate, so shouldTrace() kept evaluating a stale
predicate.
+ */
+ @SuppressWarnings("unchecked")
+ @Test
+ void testClearTraceFilter() throws Exception {
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on
+ = new ObjectName(
+ "org.apache.camel:context=" +
context.getManagementName()
+ + ",type=tracer,name=BacklogTracer");
+ assertTrue(mbeanServer.isRegistered(on));
+
+ // disable removeOnDump so we can count events across phases
+ mbeanServer.setAttribute(on, new Attribute("RemoveOnDump",
Boolean.FALSE));
+
+ // enable tracing with a filter that requires header "foo"
+ mbeanServer.setAttribute(on, new Attribute("Enabled", Boolean.TRUE));
+ mbeanServer.setAttribute(on, new Attribute("TraceFilter",
"${header.foo} != null"));
+
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+
+ // send message WITH header — should be traced
+ template.sendBodyAndHeader("direct:start", "Matched", "foo", 123);
+ assertMockEndpointsSatisfied();
+
+ List<BacklogTracerEventMessage> events
+ = (List<BacklogTracerEventMessage>) mbeanServer.invoke(on,
"dumpAllTracedMessages", null, null);
+ int tracedWithFilter = events.size();
+ assertTrue(tracedWithFilter > 0, "Message with header should be
traced");
+
+ // send message WITHOUT header — should NOT be traced due to filter
+ resetMocks();
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+ template.sendBody("direct:start", "Not Matched");
+ assertMockEndpointsSatisfied();
+
+ events = (List<BacklogTracerEventMessage>) mbeanServer.invoke(on,
"dumpAllTracedMessages", null, null);
+ assertEquals(tracedWithFilter, events.size(), "Unmatched message
should NOT add trace events");
+
+ // clear the filter
+ mbeanServer.setAttribute(on, new Attribute("TraceFilter", null));
+ String traceFilter = (String) mbeanServer.getAttribute(on,
"TraceFilter");
+ assertNull(traceFilter, "TraceFilter should be null after clearing");
+
+ // send a message WITHOUT header — should now be traced since filter
is cleared
+ resetMocks();
+ getMockEndpoint("mock:foo").expectedMessageCount(1);
+ getMockEndpoint("mock:bar").expectedMessageCount(1);
+ template.sendBody("direct:start", "After Clear");
+ assertMockEndpointsSatisfied();
+
+ events = (List<BacklogTracerEventMessage>) mbeanServer.invoke(on,
"dumpAllTracedMessages", null, null);
+ assertFalse(events.size() == tracedWithFilter,
+ "After clearing filter, additional trace events should appear
(before=" + tracedWithFilter
+ + ", after=" +
events.size() + ")");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.setUseBreadcrumb(false);
+ context.setBacklogTracingStandby(true);
+
+ from("direct:start")
+ .to("mock:foo").id("foo")
+ .to("mock:bar").id("bar");
+ }
+ };
+ }
+
+}
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..bcbfa60dfab8 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
@@ -56,14 +56,14 @@ public abstract class ScheduledPollConsumer extends
DefaultConsumer
// if adding more options then align with
org.apache.camel.support.ScheduledPollEndpoint
private boolean startScheduler = true;
- private long initialDelay = 1000;
- private long delay = 500;
- private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
- private boolean useFixedDelay = true;
+ private volatile long initialDelay = 1000;
+ private volatile long delay = 500;
+ private volatile TimeUnit timeUnit = TimeUnit.MILLISECONDS;
+ private volatile 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..9a54ef0d92c6 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,21 @@ 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;
+
+ // immutable holder for throttling limits that must be visible atomically
on routing threads
+ private record ThrottlingLimits(int maxInflightExchanges, int
resumeInflightExchanges) {
+ }
+
+ private volatile ThrottlingLimits throttlingLimits = new
ThrottlingLimits(1000, 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;
@@ -128,15 +135,20 @@ public class ThrottlingInflightRoutePolicy extends
RoutePolicySupport implements
// this works the best when this logic is executed when the exchange
is done
Consumer consumer = route.getConsumer();
+ // snapshot the volatile holder once for consistent reads
+ ThrottlingLimits limits = this.throttlingLimits;
+ int maxInflight = limits.maxInflightExchanges();
+ int resumeInflight = limits.resumeInflightExchanges();
+
int size = getSize(route, exchange);
- boolean stop = maxInflightExchanges > 0 && size > maxInflightExchanges;
+ boolean stop = maxInflight > 0 && size > maxInflight;
if (LOG.isTraceEnabled()) {
- LOG.trace("{} > 0 && {} > {} evaluated as {}",
maxInflightExchanges, size, maxInflightExchanges, stop);
+ LOG.trace("{} > 0 && {} > {} evaluated as {}", maxInflight, size,
maxInflight, stop);
}
if (stop) {
try {
lock.lock();
- stopConsumer(size, consumer);
+ stopConsumer(size, consumer, maxInflight);
} catch (Exception e) {
handleException(e);
} finally {
@@ -147,14 +159,14 @@ public class ThrottlingInflightRoutePolicy extends
RoutePolicySupport implements
// reload size in case a race condition with too many at once being
invoked
// so we need to ensure that we read the most current size and start
the consumer if we are already to low
size = getSize(route, exchange);
- boolean start = size <= resumeInflightExchanges;
+ boolean start = size <= resumeInflight;
if (LOG.isTraceEnabled()) {
- LOG.trace("{} <= {} evaluated as {}", size,
resumeInflightExchanges, start);
+ LOG.trace("{} <= {} evaluated as {}", size, resumeInflight, start);
}
if (start) {
try {
lock.lock();
- startConsumer(size, consumer);
+ startConsumer(size, consumer, resumeInflight);
} catch (Exception e) {
handleException(e);
} finally {
@@ -178,7 +190,10 @@ public class ThrottlingInflightRoutePolicy extends
RoutePolicySupport implements
public void setMaxInflightExchanges(int maxInflightExchanges) {
this.maxInflightExchanges = maxInflightExchanges;
// recalculate, must be at least at 1
- this.resumeInflightExchanges = Math.max(resumePercentOfMax *
maxInflightExchanges / 100, 1);
+ int resume = Math.max(resumePercentOfMax * maxInflightExchanges / 100,
1);
+ this.resumeInflightExchanges = resume;
+ // atomically publish both values for routing threads
+ this.throttlingLimits = new ThrottlingLimits(maxInflightExchanges,
resume);
}
public int getResumePercentOfMax() {
@@ -199,7 +214,10 @@ public class ThrottlingInflightRoutePolicy extends
RoutePolicySupport implements
this.resumePercentOfMax = resumePercentOfMax;
// recalculate, must be at least at 1
- this.resumeInflightExchanges = Math.max(resumePercentOfMax *
maxInflightExchanges / 100, 1);
+ int resume = Math.max(resumePercentOfMax * maxInflightExchanges / 100,
1);
+ this.resumeInflightExchanges = resume;
+ // atomically publish both values for routing threads
+ this.throttlingLimits = new ThrottlingLimits(maxInflightExchanges,
resume);
}
public ThrottlingScope getScope() {
@@ -258,18 +276,18 @@ public class ThrottlingInflightRoutePolicy extends
RoutePolicySupport implements
}
}
- private void startConsumer(int size, Consumer consumer) throws Exception {
+ private void startConsumer(int size, Consumer consumer, int
resumeInflight) throws Exception {
boolean started = resumeOrStartConsumer(consumer);
if (started) {
- getLogger().log("Throttling consumer: " + size + " <= " +
resumeInflightExchanges
+ getLogger().log("Throttling consumer: " + size + " <= " +
resumeInflight
+ " inflight exchange by resuming consumer: " +
consumer);
}
}
- private void stopConsumer(int size, Consumer consumer) throws Exception {
+ private void stopConsumer(int size, Consumer consumer, int maxInflight)
throws Exception {
boolean stopped = suspendOrStopConsumer(consumer);
if (stopped) {
- getLogger().log("Throttling consumer: " + size + " > " +
maxInflightExchanges
+ getLogger().log("Throttling consumer: " + size + " > " +
maxInflight
+ " inflight exchange by suspending consumer: " +
consumer);
}
}
