This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch activemq-6.2.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-6.2.x by this push:
new a919915797 [6.2.x] Fix TextMessage body thread safety (#2188) (#2204)
a919915797 is described below
commit a919915797228936b4308327b0288058f4dd337e
Author: Christopher L. Shannon <[email protected]>
AuthorDate: Thu Jul 9 19:45:07 2026 -0400
[6.2.x] Fix TextMessage body thread safety (#2188) (#2204)
* Fix TextMessage body thread safety (#2188)
This fixes the rare null body issues seen on the broker when using text
messages by using synchronization when marshaling or unmarshaling the
body into a String to prevent a race condition from causing the body to
end up null. This has been optimized to use volatiles and double checked
locking to avoid synchronization unless needed.
(cherry picked from commit b131c3d2661ad28236a5ac18826e7d88cd3cb976)
* 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]>
(cherry picked from commit e1368d5f45570f5667d73cd1c9fd7ce066fe6cff)
---------
Co-authored-by: Matt Pavlovich <[email protected]>
---
.../activemq/command/ActiveMQTextMessage.java | 124 ++++++---
.../java/org/apache/activemq/command/Message.java | 8 +-
.../ActiveMQTextMessageContentRaceTest.java | 306 +++++++++++++++++++++
3 files changed, 396 insertions(+), 42 deletions(-)
diff --git
a/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java
b/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java
index a5df4164f0..0ace802c8c 100644
---
a/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java
+++
b/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java
@@ -45,20 +45,19 @@ public class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
public static final byte DATA_STRUCTURE_TYPE =
CommandTypes.ACTIVEMQ_TEXT_MESSAGE;
- protected String text;
+ // This is package scope (instead of private) for testing purposes
+ volatile String text;
@Override
public Message copy() {
ActiveMQTextMessage copy = new ActiveMQTextMessage();
- copy(copy);
+ synchronized (this) {
+ super.copy(copy);
+ copy.text = text;
+ }
return copy;
}
- private void copy(ActiveMQTextMessage copy) {
- super.copy(copy);
- copy.text = text;
- }
-
@Override
public byte getDataStructureType() {
return DATA_STRUCTURE_TYPE;
@@ -72,19 +71,35 @@ public class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
@Override
public void setText(String text) throws MessageNotWriteableException {
checkReadOnlyBody();
- this.text = text;
- setContent(null);
+ synchronized (this) {
+ this.text = text;
+ setContent(null);
+ }
+ }
+
+ // Synchronize this to prevent setting content if another mutation
operation
+ // is happening concurrently.
+ @Override
+ public synchronized void setContent(ByteSequence content) {
+ super.setContent(content);
}
@Override
public String getText() throws JMSException {
- ByteSequence content = getContent();
+ String text = this.text;
- if (text == null && content != null) {
- text = decodeContent(content);
- setContent(null);
- setCompressed(false);
+ if (text == null) {
+ synchronized (this) {
+ text = this.text;
+ // Double-checked locking, re-check under lock if we need to
decode
+ if (text == null && content != null) {
+ this.text = text = decodeContent(content);
+ setContent(null);
+ setCompressed(false);
+ }
+ }
}
+
return text;
}
@@ -124,27 +139,36 @@ public class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
@Override
public void storeContentAndClear() {
- storeContent();
- text=null;
+ // always lock to simplify things because if this method is being
called
+ // it's right before send so it's very likely to need to mutate state
+ // This should generally be uncontested lock so it will be fast
+ synchronized (this) {
+ storeContent();
+ text = null;
+ }
}
@Override
public void storeContent() {
try {
- ByteSequence content = getContent();
- String text = this.text;
- if (content == null && text != null) {
- ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
- OutputStream os = bytesOut;
- ActiveMQConnection connection = getConnection();
- if (connection != null && connection.isUseCompression()) {
- compressed = true;
- os = new DeflaterOutputStream(os);
+ // Content is volatile so if it's not null we can skip and do
nothing
+ if (content == null) {
+ synchronized (this) {
+ // Double-checked locking, re-check state under lock
+ if (content == null && text != null) {
+ ByteArrayOutputStream bytesOut = new
ByteArrayOutputStream();
+ OutputStream os = bytesOut;
+ ActiveMQConnection connection = getConnection();
+ if (connection != null &&
connection.isUseCompression()) {
+ compressed = true;
+ os = new DeflaterOutputStream(os);
+ }
+ DataOutputStream dataOut = new DataOutputStream(os);
+ MarshallingSupport.writeUTF8(dataOut, text);
+ dataOut.close();
+ setContent(bytesOut.toByteSequence());
+ }
}
- DataOutputStream dataOut = new DataOutputStream(os);
- MarshallingSupport.writeUTF8(dataOut, text);
- dataOut.close();
- setContent(bytesOut.toByteSequence());
}
} catch (IOException e) {
throw new RuntimeException(e);
@@ -156,11 +180,19 @@ public class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
@Override
public void clearUnMarshalledState() throws JMSException {
super.clearUnMarshalledState();
- this.text = null;
+ if (this.text != null) {
+ synchronized (this) {
+ // This is volatile but another locking makes sure another
thread
+ // isn't attempting to read this value to marshal to the
content at the
+ // same time
+ this.text = null;
+ }
+ }
}
+ // We need to sync because both variables need to be read independently
@Override
- public boolean isContentMarshalled() {
+ public synchronized boolean isContentMarshalled() {
return content != null || text == null;
}
@@ -177,19 +209,31 @@ public class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
*/
@Override
public void clearBody() throws JMSException {
- super.clearBody();
- this.text = null;
+ synchronized (this) {
+ super.clearBody();
+ this.text = null;
+ }
}
@Override
public int getSize() {
- String text = this.text;
- if (size == 0 && content == null && text != null) {
- size = getMinimumMessageSize();
- if (marshalledProperties != null) {
- size += marshalledProperties.getLength();
+ int size = this.size;
+ if (size == 0) {
+ synchronized (this) {
+ size = this.size;
+ String text = this.text;
+ ByteSequence content = getContent();
+ if (size == 0 && content == null && text != null) {
+ size = getMinimumMessageSize();
+ ByteSequence marshalledProperties =
this.marshalledProperties;
+ if (marshalledProperties != null) {
+ size += marshalledProperties.getLength();
+ }
+ size += text.length() * 2;
+ this.size = size;
+ }
+ return super.getSize();
}
- size += text.length() * 2;
}
return super.getSize();
}
@@ -198,7 +242,7 @@ public class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
public String toString() {
try {
String text = this.text;
- if( text == null ) {
+ if (text == null) {
text = decodeContent(getContent());
}
if (text != null) {
diff --git
a/activemq-client/src/main/java/org/apache/activemq/command/Message.java
b/activemq-client/src/main/java/org/apache/activemq/command/Message.java
index 06e0aebeea..11ccd306fa 100644
--- a/activemq-client/src/main/java/org/apache/activemq/command/Message.java
+++ b/activemq-client/src/main/java/org/apache/activemq/command/Message.java
@@ -81,12 +81,12 @@ public abstract class Message extends BaseCommand
implements MarshallAware, Mess
protected boolean compressed;
protected String userID;
- protected ByteSequence content;
+ protected volatile ByteSequence content;
protected volatile ByteSequence marshalledProperties;
protected DataStructure dataStructure;
protected int redeliveryCounter;
- protected int size;
+ protected volatile int size;
protected Map<String, Object> properties;
protected boolean readOnlyProperties;
protected boolean readOnlyBody;
@@ -715,14 +715,18 @@ public abstract class Message extends BaseCommand
implements MarshallAware, Mess
@Override
public int getSize() {
int minimumMessageSize = getMinimumMessageSize();
+ ByteSequence content = this.content;
+ int size = this.size;
if (size < minimumMessageSize || size == 0) {
size = minimumMessageSize;
+ ByteSequence marshalledProperties = this.marshalledProperties;
if (marshalledProperties != null) {
size += marshalledProperties.getLength();
}
if (content != null) {
size += content.getLength();
}
+ this.size = size;
}
return size;
}
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
