This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/main by this push:
new e1368d5f45 Add tests validating TextMessage body thread safety fix
(#2207)
e1368d5f45 is described below
commit e1368d5f45570f5667d73cd1c9fd7ce066fe6cff
Author: Christopher L. Shannon <[email protected]>
AuthorDate: Thu Jul 9 17:11:14 2026 -0400
Add tests validating TextMessage body thread safety fix (#2207)
Validates the fix for the race between getText(), storeContentAndClear()
and copy() on a shared ActiveMQTextMessage instance that could null the
message body: getText() decoded content into text and cleared content
while storeContentAndClear() observed the pre-clear content, skipped
re-encoding, and cleared text — losing the body permanently. Triggered on
the broker by XPath selector evaluation or JMX browse racing dispatch
marshalling, with the network bridge copy() adding a third participant.
Four tests:
- sequential control: getText/storeContentAndClear round trips preserve
the body
- two-way race: 20k iterations of concurrent getText vs
storeContentAndClear; no iteration may lose the body
- three-way race including copy(): both the original and the copy must
retain a recoverable body (pre-fix this loses the original's body
within a few thousand iterations)
- deterministic lock-discipline pin: the fix serializes the text/content
state transitions on the message instance, so storeContentAndClear()
must block while another thread holds the message monitor; fails
immediately on any regression to unsynchronized transitions
Verified failing (2 of 4) against the pre-fix base and passing on this
branch.
Co-authored-by: Matt Pavlovich <[email protected]>
---
.../ActiveMQTextMessageContentRaceTest.java | 306 +++++++++++++++++++++
1 file changed, 306 insertions(+)
diff --git
a/activemq-client/src/test/java/org/apache/activemq/command/ActiveMQTextMessageContentRaceTest.java
b/activemq-client/src/test/java/org/apache/activemq/command/ActiveMQTextMessageContentRaceTest.java
new file mode 100644
index 0000000000..698eb43c69
--- /dev/null
+++
b/activemq-client/src/test/java/org/apache/activemq/command/ActiveMQTextMessageContentRaceTest.java
@@ -0,0 +1,306 @@
+/**
+ * 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.activemq.command;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Validates the fix for the race condition between getText(),
+ * storeContentAndClear() and copy() in ActiveMQTextMessage that could cause
+ * the message body to become permanently null.
+ *
+ * The race existed because those methods performed multi-step
+ * read-modify-clear operations on two fields (text and content) without a
+ * common monitor:
+ *
+ * Thread A (getText): Thread B (storeContentAndClear):
+ * 1. reads content -> non-null 1. storeContent(): content non-null,
+ * 2. decodes content -> text skips encoding text -> content
+ * 3. setContent(null) 2. text = null
+ *
+ * Result: content=null (cleared by A), text=null (cleared by B).
+ * The body was permanently lost.
+ *
+ * Real-world trigger: on the broker side, a transport thread calls
+ * beforeMarshall() -> storeContentAndClear() while concurrently an XPath
+ * selector evaluator (JAXPXPathEvaluator) or JMX browse calls getText() on
+ * the same Message instance, and the network bridge calls copy() on it. The
+ * broker dispatches the same Message object to multiple consumers without
+ * copying it.
+ *
+ * The fix synchronizes the state transitions on the message instance
+ * itself: getText() decodes+clears under {@code synchronized(this)} (with a
+ * double-checked volatile read of text for the fast path),
+ * storeContentAndClear() stores+clears under the same monitor, copy() takes
+ * the monitor while snapshotting both fields, and content/text are volatile.
+ */
+public class ActiveMQTextMessageContentRaceTest {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ActiveMQTextMessageContentRaceTest.class);
+
+ private static final String TEST_TEXT = "Hello, World! This is a test
message body.";
+ private static final int ITERATIONS = 20_000;
+
+ /**
+ * Creates a message in the state it has on the broker after arriving and
+ * being trimmed by reduceMemoryFootprint: content is the marshalled
+ * bytes, text is null.
+ */
+ private ActiveMQTextMessage brokerStateMessage() throws Exception {
+ ActiveMQTextMessage msg = new ActiveMQTextMessage();
+ msg.setText(TEST_TEXT);
+ msg.storeContent();
+ msg.clearUnMarshalledState();
+ return msg;
+ }
+
+ /**
+ * Control case: sequential getText() / storeContentAndClear() round
+ * trips preserve the body.
+ */
+ @Test
+ public void testSequentialGetTextAndStoreContentPreservesBody() throws
Exception {
+ ActiveMQTextMessage msg = brokerStateMessage();
+ assertNotNull("content should be set after storeContent()",
msg.getContent());
+
+ // getText() decodes content -> text (and clears content)
+ assertEquals(TEST_TEXT, msg.getText());
+
+ // storeContentAndClear() re-encodes text -> content, clears text
+ msg.storeContentAndClear();
+ assertNotNull("content should be set after storeContentAndClear()",
msg.getContent());
+
+ // getText() again recovers the body
+ assertEquals(TEST_TEXT, msg.getText());
+ }
+
+ /**
+ * The original race: concurrent getText() and storeContentAndClear()
+ * on the same instance. Before the fix this lost the body within a few
+ * thousand iterations on any multi-core machine; with the state
+ * transitions synchronized on the message instance, no iteration may
+ * lose the body.
+ */
+ @Test(timeout = 120_000)
+ public void testConcurrentGetTextAndStoreContentAndClearPreservesBody()
throws Exception {
+ final AtomicInteger nullBodyCount = new AtomicInteger(0);
+ final AtomicReference<String> firstFailure = new AtomicReference<>();
+
+ for (int i = 0; i < ITERATIONS; i++) {
+ ActiveMQTextMessage msg = brokerStateMessage();
+
+ CyclicBarrier barrier = new CyclicBarrier(2);
+
+ // simulates XPath selector evaluation / JMX browse
+ Thread readerThread = new Thread(() -> {
+ try {
+ barrier.await();
+ msg.getText();
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+
+ // simulates the transport thread marshalling for dispatch
+ Thread marshalThread = new Thread(() -> {
+ try {
+ barrier.await();
+ msg.storeContentAndClear();
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+
+ readerThread.start();
+ marshalThread.start();
+ readerThread.join();
+ marshalThread.join();
+
+ String recoveredText = null;
+ try {
+ recoveredText = msg.getText();
+ } catch (Exception e) {
+ // getText might throw if content is corrupted
+ }
+
+ if (recoveredText == null) {
+ int count = nullBodyCount.incrementAndGet();
+ if (firstFailure.get() == null) {
+ firstFailure.set("Iteration " + i + ": text=" + msg.text +
+ ", content=" + msg.getContent());
+ }
+ if (count >= 10) {
+ break;
+ }
+ }
+ }
+
+ if (nullBodyCount.get() > 0) {
+ fail("RACE CONDITION: " + nullBodyCount.get() + " messages had
null body due to " +
+ "concurrent getText() and storeContentAndClear(). First
failure: " +
+ firstFailure.get());
+ }
+ LOG.info("Body preserved across {} concurrent
getText/storeContentAndClear iterations",
+ ITERATIONS);
+ }
+
+ /**
+ * Three-way race including copy(): the network bridge copies the same
+ * Message instance that local dispatch is marshalling while a selector
+ * reads it. Both the original and the copy must retain a recoverable
+ * body.
+ */
+ @Test(timeout = 120_000)
+ public void testConcurrentCopyGetTextAndStoreContentPreservesBody() throws
Exception {
+ final AtomicInteger failures = new AtomicInteger(0);
+ final AtomicReference<String> firstFailure = new AtomicReference<>();
+
+ for (int i = 0; i < ITERATIONS / 2; i++) {
+ ActiveMQTextMessage msg = brokerStateMessage();
+ AtomicReference<Message> copyRef = new AtomicReference<>();
+
+ CyclicBarrier barrier = new CyclicBarrier(3);
+
+ Thread readerThread = new Thread(() -> {
+ try {
+ barrier.await();
+ msg.getText();
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+ Thread marshalThread = new Thread(() -> {
+ try {
+ barrier.await();
+ msg.storeContentAndClear();
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+ Thread copyThread = new Thread(() -> {
+ try {
+ barrier.await();
+ copyRef.set(msg.copy());
+ } catch (Exception e) {
+ // ignore
+ }
+ });
+
+ readerThread.start();
+ marshalThread.start();
+ copyThread.start();
+ readerThread.join();
+ marshalThread.join();
+ copyThread.join();
+
+ String originalText = null;
+ String copyText = null;
+ try {
+ originalText = msg.getText();
+ ActiveMQTextMessage copy = (ActiveMQTextMessage) copyRef.get();
+ copyText = copy != null ? copy.getText() : "no-copy-made";
+ } catch (Exception e) {
+ // fall through with nulls
+ }
+
+ if (originalText == null || copyText == null) {
+ int count = failures.incrementAndGet();
+ if (firstFailure.get() == null) {
+ firstFailure.set("Iteration " + i + ": original=" +
originalText +
+ ", copy=" + copyText);
+ }
+ if (count >= 10) {
+ break;
+ }
+ }
+ }
+
+ if (failures.get() > 0) {
+ fail("RACE CONDITION: " + failures.get() + " iterations lost the
body on the " +
+ "original or its copy() during concurrent
copy/getText/storeContentAndClear. " +
+ "First failure: " + firstFailure.get());
+ }
+ LOG.info("Body preserved on original and copy across {} three-way
iterations",
+ ITERATIONS / 2);
+ }
+
+ /**
+ * Deterministically pins the locking discipline of the fix: the state
+ * transitions synchronize on the message instance itself. While another
+ * thread holds the message monitor, storeContentAndClear() must block —
+ * so the getText() decode+clear and the storeContentAndClear()
+ * store+clear can never interleave.
+ *
+ * If the implementation ever moves off {@code synchronized(this)} (or
+ * drops synchronization), this test fails and the concurrent tests above
+ * become the (probabilistic) safety net.
+ */
+ @Test(timeout = 60_000)
+ public void testStateTransitionsSynchronizeOnMessageInstance() throws
Exception {
+ ActiveMQTextMessage msg = brokerStateMessage();
+ // decode so text is populated and storeContentAndClear has work to do
+ assertEquals(TEST_TEXT, msg.getText());
+
+ CountDownLatch monitorHeld = new CountDownLatch(1);
+ CountDownLatch releaseMonitor = new CountDownLatch(1);
+ Thread holder = new Thread(() -> {
+ synchronized (msg) {
+ monitorHeld.countDown();
+ try {
+ releaseMonitor.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ holder.start();
+ assertTrue("holder should acquire the message monitor",
monitorHeld.await(5, TimeUnit.SECONDS));
+
+ CountDownLatch storeDone = new CountDownLatch(1);
+ Thread storer = new Thread(() -> {
+ msg.storeContentAndClear();
+ storeDone.countDown();
+ });
+ storer.start();
+
+ assertTrue("storeContentAndClear must BLOCK while the message monitor
is held — " +
+ "the fix serializes text/content state transitions on the
message instance",
+ !storeDone.await(500, TimeUnit.MILLISECONDS));
+
+ releaseMonitor.countDown();
+ assertTrue("storeContentAndClear should complete once the monitor is
released",
+ storeDone.await(5, TimeUnit.SECONDS));
+ holder.join(5000);
+ storer.join(5000);
+
+ assertNotNull("content must be set after storeContentAndClear",
msg.getContent());
+ assertEquals("body must survive the round trip", TEST_TEXT,
msg.getText());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
