Skip to content

[6.2.x] Minor updates to text message sync (#2209) #2214

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cshannon merged 1 commit into from
Jul 10, 2026
+170 −51
Merged

[6.2.x] Minor updates to text message sync (#2209) #2214

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 24 additions & 27 deletions activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,13 +68,17 @@ public String getJMSXMimeType() {
return "jms/text-message";
}

// add sync
@Override
public void setText(String text) throws MessageNotWriteableException {
public synchronized void compress() throws IOException {
super.compress();
}

@Override
public synchronized void setText(String text) throws MessageNotWriteableException {
checkReadOnlyBody();
synchronized (this) {
this.text = text;
setContent(null);
}
this.text = text;
setContent(null);
}

// Synchronize this to prevent setting content if another mutation operation
Expand Down Expand Up @@ -137,15 +141,13 @@ public void beforeMarshall(WireFormat wireFormat) throws IOException {
storeContentAndClear();
}

// 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
@Override
public void storeContentAndClear() {
// 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;
}
public synchronized void storeContentAndClear() {
storeContent();
text = null;
}

@Override
Expand Down Expand Up @@ -178,15 +180,12 @@ public void storeContent() {
// see https://issues.apache.org/activemq/browse/AMQ-2103
// and https://issues.apache.org/activemq/browse/AMQ-2966
@Override
public void clearUnMarshalledState() throws JMSException {
super.clearUnMarshalledState();
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;
}
public synchronized void clearUnMarshalledState() throws JMSException {
// Double check this under lock, we should only clear if
// the text/properties are marshaled
if (isMarshalled()) {
super.clearUnMarshalledState();
this.text = null;
}
}

Expand All @@ -208,11 +207,9 @@ public synchronized boolean isContentMarshalled() {
* due to some internal error.
*/
@Override
public void clearBody() throws JMSException {
synchronized (this) {
super.clearBody();
this.text = null;
}
public synchronized void clearBody() throws JMSException {
super.clearBody();
this.text = null;
}

@Override
Expand Down
170 changes: 146 additions & 24 deletions ...d/ActiveMQTextMessageContentRaceTest.java → ...mmand/ActiveMQMessageContentRaceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,61 +21,117 @@
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import jakarta.jms.JMSException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Map;
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.Assume;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
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.
*
* <p>
* The race existed because those methods performed multi-step
* read-modify-clear operations on two fields (text and content) without a
* common monitor:
*
* <p>
* 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.
*
* <p>
* 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.
*
* <p>
* 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.
* <p>
* Note:
* This issue is specific to ActiveMQTextMessage because it is the only message
* type that will only store either the unmarshalled data (text) or the
* bytes, but not both. Other message types do not clear the content after unmarshaling.
* This means that if multiple threads try and do a conversion on a message like Map
* or Object message, it may convert twice, but you won't get null. The other
* message types only clear unmarshaled content at certain points in the broker
* if reduceMemoryFootprint is true.
* <p>
* This test also runs the relevant race condition tests on other message types
* to confirm they do not suffer from the same null body issue.
*/
public class ActiveMQTextMessageContentRaceTest {
@RunWith(Parameterized.class)
public class ActiveMQMessageContentRaceTest {

private static final Logger LOG = LoggerFactory.getLogger(ActiveMQTextMessageContentRaceTest.class);
private static final Logger LOG = LoggerFactory.getLogger(ActiveMQMessageContentRaceTest.class);

private static final String TEST_TEXT = "Hello, World! This is a test message body.";
private static final int ITERATIONS = 20_000;
private final byte messageType;

public ActiveMQMessageContentRaceTest(byte messageType) {
this.messageType = messageType;
}

@Parameters
public static Collection<Object[]> data() {
return Arrays.asList(new Object[][] {
{ActiveMQTextMessage.DATA_STRUCTURE_TYPE},
// Add other message types to verify they don't suffer
// from the same race condition either
{ActiveMQBytesMessage.DATA_STRUCTURE_TYPE},
{ActiveMQStreamMessage.DATA_STRUCTURE_TYPE},
{ActiveMQMapMessage.DATA_STRUCTURE_TYPE},
{ActiveMQObjectMessage.DATA_STRUCTURE_TYPE},
});
}

/**
* 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);
private ActiveMQMessage brokerStateMessage() throws Exception {
final ActiveMQMessage msg;
if (messageType == ActiveMQTextMessage.DATA_STRUCTURE_TYPE) {
msg = new ActiveMQTextMessage();
asText(msg).setText(TEST_TEXT);
} else if (messageType == ActiveMQBytesMessage.DATA_STRUCTURE_TYPE) {
msg = new ActiveMQBytesMessage();
asBytes(msg).writeBytes(TEST_TEXT.getBytes());
} else if (messageType == ActiveMQStreamMessage.DATA_STRUCTURE_TYPE) {
msg = new ActiveMQStreamMessage();
asStream(msg).writeBytes(TEST_TEXT.getBytes());
} else if (messageType == ActiveMQMapMessage.DATA_STRUCTURE_TYPE) {
msg = new ActiveMQMapMessage();
asMap(msg).setString("test1", TEST_TEXT);
} else if (messageType == ActiveMQObjectMessage.DATA_STRUCTURE_TYPE) {
msg = new ActiveMQObjectMessage();
asObject(msg).setObject(TEST_TEXT);
} else {
throw new IllegalArgumentException("Unsupported data structure type: " + messageType);
}
msg.storeContent();
msg.clearUnMarshalledState();
return msg;
Expand All @@ -87,18 +143,18 @@ private ActiveMQTextMessage brokerStateMessage() throws Exception {
*/
@Test
public void testSequentialGetTextAndStoreContentPreservesBody() throws Exception {
ActiveMQTextMessage msg = brokerStateMessage();
ActiveMQMessage msg = brokerStateMessage();
assertNotNull("content should be set after storeContent()", msg.getContent());

// getText() decodes content -> text (and clears content)
assertEquals(TEST_TEXT, msg.getText());
assertEquals(TEST_TEXT, decodeBody(msg));

// 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());
assertEquals(TEST_TEXT, decodeBody(msg));
}

/**
Expand All @@ -114,15 +170,15 @@ public void testConcurrentGetTextAndStoreContentAndClearPreservesBody() throws E
final AtomicReference<String> firstFailure = new AtomicReference<>();

for (int i = 0; i < ITERATIONS; i++) {
ActiveMQTextMessage msg = brokerStateMessage();
ActiveMQMessage msg = brokerStateMessage();

CyclicBarrier barrier = new CyclicBarrier(2);

// simulates XPath selector evaluation / JMX browse
Thread readerThread = new Thread(() -> {
try {
barrier.await();
msg.getText();
decodeBody(msg);
} catch (Exception e) {
// ignore
}
Expand All @@ -145,15 +201,15 @@ public void testConcurrentGetTextAndStoreContentAndClearPreservesBody() throws E

String recoveredText = null;
try {
recoveredText = msg.getText();
recoveredText = decodeBody(msg);
} 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 +
firstFailure.set("Iteration " + i + ": text=" + rawBody(msg) +
", content=" + msg.getContent());
}
if (count >= 10) {
Expand Down Expand Up @@ -183,15 +239,15 @@ public void testConcurrentCopyGetTextAndStoreContentPreservesBody() throws Excep
final AtomicReference<String> firstFailure = new AtomicReference<>();

for (int i = 0; i < ITERATIONS / 2; i++) {
ActiveMQTextMessage msg = brokerStateMessage();
ActiveMQMessage msg = brokerStateMessage();
AtomicReference<Message> copyRef = new AtomicReference<>();

CyclicBarrier barrier = new CyclicBarrier(3);

Thread readerThread = new Thread(() -> {
try {
barrier.await();
msg.getText();
decodeBody(msg);
} catch (Exception e) {
// ignore
}
Expand Down Expand Up @@ -223,9 +279,9 @@ public void testConcurrentCopyGetTextAndStoreContentPreservesBody() throws Excep
String originalText = null;
String copyText = null;
try {
originalText = msg.getText();
ActiveMQTextMessage copy = (ActiveMQTextMessage) copyRef.get();
copyText = copy != null ? copy.getText() : "no-copy-made";
originalText = decodeBody(msg);
ActiveMQMessage copy = (ActiveMQMessage) copyRef.get();
copyText = copy != null ? decodeBody(copy): "no-copy-made";
} catch (Exception e) {
// fall through with nulls
}
Expand Down Expand Up @@ -264,9 +320,12 @@ public void testConcurrentCopyGetTextAndStoreContentPreservesBody() throws Excep
*/
@Test(timeout = 60_000)
public void testStateTransitionsSynchronizeOnMessageInstance() throws Exception {
ActiveMQTextMessage msg = brokerStateMessage();
// Only applies to text message
Assume.assumeTrue(this.messageType == ActiveMQTextMessage.DATA_STRUCTURE_TYPE);

ActiveMQMessage msg = brokerStateMessage();
// decode so text is populated and storeContentAndClear has work to do
assertEquals(TEST_TEXT, msg.getText());
assertEquals(TEST_TEXT, decodeBody(msg));

CountDownLatch monitorHeld = new CountDownLatch(1);
CountDownLatch releaseMonitor = new CountDownLatch(1);
Expand Down Expand Up @@ -301,6 +360,69 @@ public void testStateTransitionsSynchronizeOnMessageInstance() throws Exception
storer.join(5000);

assertNotNull("content must be set after storeContentAndClear", msg.getContent());
assertEquals("body must survive the round trip", TEST_TEXT, msg.getText());
assertEquals("body must survive the round trip", TEST_TEXT, decodeBody(msg));
}

private String decodeBody(final ActiveMQMessage msg) throws JMSException {
if (messageType == ActiveMQTextMessage.DATA_STRUCTURE_TYPE) {
return asText(msg).getText();
} else if (messageType == ActiveMQBytesMessage.DATA_STRUCTURE_TYPE) {
var bytesMsg = asBytes(msg);
bytesMsg.setReadOnlyBody(true);
bytesMsg.reset();
byte[] bytes = new byte[(int) bytesMsg.getBodyLength()];
bytesMsg.readBytes(bytes);
return new String(bytes);
} else if (messageType == ActiveMQStreamMessage.DATA_STRUCTURE_TYPE) {
var bytesMsg = asStream(msg);
bytesMsg.setReadOnlyBody(true);
bytesMsg.reset();
byte[] bytes = new byte[TEST_TEXT.length()];
bytesMsg.readBytes(bytes);
return new String(bytes);
} else if (messageType == ActiveMQMapMessage.DATA_STRUCTURE_TYPE) {
return asMap(msg).getString("test1");
} else if (messageType == ActiveMQObjectMessage.DATA_STRUCTURE_TYPE) {
return asObject(msg).getObject().toString();
} else {
throw new IllegalArgumentException("Unsupported data structure type: " + messageType);
}
}

private Object rawBody(final ActiveMQMessage msg) {
if (messageType == ActiveMQTextMessage.DATA_STRUCTURE_TYPE) {
return asText(msg).text;
} else if (messageType == ActiveMQBytesMessage.DATA_STRUCTURE_TYPE) {
return asBytes(msg).content != null ? new String(asBytes(msg).content.data) : null;
} else if (messageType == ActiveMQStreamMessage.DATA_STRUCTURE_TYPE) {
return asStream(msg).content != null ? new String(asStream(msg).content.data) : null;
} else if (messageType == ActiveMQMapMessage.DATA_STRUCTURE_TYPE) {
Map<String, Object> map = asMap(msg).map;
return map != null ? map.get("test1") : null;
} else if (messageType == ActiveMQObjectMessage.DATA_STRUCTURE_TYPE) {
return asObject(msg).object;
} else {
throw new IllegalArgumentException("Unsupported data structure type: " + messageType);
}
}

private static ActiveMQTextMessage asText(ActiveMQMessage message) {
return (ActiveMQTextMessage) message;
}

private static ActiveMQBytesMessage asBytes(ActiveMQMessage message) {
return (ActiveMQBytesMessage) message;
}

private static ActiveMQStreamMessage asStream(ActiveMQMessage message) {
return (ActiveMQStreamMessage) message;
}

private static ActiveMQMapMessage asMap(ActiveMQMessage message) {
return (ActiveMQMapMessage) message;
}

private static ActiveMQObjectMessage asObject(ActiveMQMessage message) {
return (ActiveMQObjectMessage) message;
}
}
Loading