From 8120e929df378552ba4265fd123c11f9cdf0f941 Mon Sep 17 00:00:00 2001 From: "Christopher L. Shannon" Date: Fri, 10 Jul 2026 08:07:27 -0400 Subject: [PATCH] Minor updates to text message sync (#2209) This is a small follow on to #2188 to simply some of the sync by moving to the method level and also adding sync in a spot or 2 that got missed. This commit also updates the race condition test to add all the other message types to show that they do not suffer from the same race condition with null bodies either. (cherry picked from commit 04b5193747bc7e5aea15fd32b617ef36f44fec71) --- .../activemq/command/ActiveMQTextMessage.java | 51 +++--- ...va => ActiveMQMessageContentRaceTest.java} | 170 +++++++++++++++--- 2 files changed, 170 insertions(+), 51 deletions(-) rename activemq-client/src/test/java/org/apache/activemq/command/{ActiveMQTextMessageContentRaceTest.java => ActiveMQMessageContentRaceTest.java} (63%) 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 0ace802c8ce..aa5a38989a7 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 @@ -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 @@ -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 @@ -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; } } @@ -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 diff --git a/activemq-client/src/test/java/org/apache/activemq/command/ActiveMQTextMessageContentRaceTest.java b/activemq-client/src/test/java/org/apache/activemq/command/ActiveMQMessageContentRaceTest.java similarity index 63% rename from activemq-client/src/test/java/org/apache/activemq/command/ActiveMQTextMessageContentRaceTest.java rename to activemq-client/src/test/java/org/apache/activemq/command/ActiveMQMessageContentRaceTest.java index 698eb43c69d..9e2b10b4206 100644 --- a/activemq-client/src/test/java/org/apache/activemq/command/ActiveMQTextMessageContentRaceTest.java +++ b/activemq-client/src/test/java/org/apache/activemq/command/ActiveMQMessageContentRaceTest.java @@ -21,13 +21,21 @@ 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; @@ -35,11 +43,11 @@ * 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 @@ -47,35 +55,83 @@ * * 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. + *

+ * 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. + *

+ * 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 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; @@ -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)); } /** @@ -114,7 +170,7 @@ public void testConcurrentGetTextAndStoreContentAndClearPreservesBody() throws E final AtomicReference firstFailure = new AtomicReference<>(); for (int i = 0; i < ITERATIONS; i++) { - ActiveMQTextMessage msg = brokerStateMessage(); + ActiveMQMessage msg = brokerStateMessage(); CyclicBarrier barrier = new CyclicBarrier(2); @@ -122,7 +178,7 @@ public void testConcurrentGetTextAndStoreContentAndClearPreservesBody() throws E Thread readerThread = new Thread(() -> { try { barrier.await(); - msg.getText(); + decodeBody(msg); } catch (Exception e) { // ignore } @@ -145,7 +201,7 @@ 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 } @@ -153,7 +209,7 @@ public void testConcurrentGetTextAndStoreContentAndClearPreservesBody() throws E 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) { @@ -183,7 +239,7 @@ public void testConcurrentCopyGetTextAndStoreContentPreservesBody() throws Excep final AtomicReference firstFailure = new AtomicReference<>(); for (int i = 0; i < ITERATIONS / 2; i++) { - ActiveMQTextMessage msg = brokerStateMessage(); + ActiveMQMessage msg = brokerStateMessage(); AtomicReference copyRef = new AtomicReference<>(); CyclicBarrier barrier = new CyclicBarrier(3); @@ -191,7 +247,7 @@ public void testConcurrentCopyGetTextAndStoreContentPreservesBody() throws Excep Thread readerThread = new Thread(() -> { try { barrier.await(); - msg.getText(); + decodeBody(msg); } catch (Exception e) { // ignore } @@ -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 } @@ -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); @@ -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 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; } }