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 81635e9974 Minor updates to text message sync (#2209) (#2214)
81635e9974 is described below
commit 81635e99742873beba32d11b2419ccaabde92f2c
Author: Christopher L. Shannon <[email protected]>
AuthorDate: Fri Jul 10 09:12:55 2026 -0400
Minor updates to text message sync (#2209) (#2214)
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 +++----
...st.java => ActiveMQMessageContentRaceTest.java} | 170 ++++++++++++++++++---
2 files changed, 170 insertions(+), 51 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 0ace802c8c..aa5a38989a 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 class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
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 class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
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 class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
// 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 class ActiveMQTextMessage extends ActiveMQMessage
implements TextMessage
* 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 698eb43c69..9e2b10b420 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.assertNotNull;
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 @@ 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
@@ -47,35 +55,83 @@ import org.slf4j.LoggerFactory;
*
* 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;
@@ -87,18 +143,18 @@ public class ActiveMQTextMessageContentRaceTest {
*/
@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 class ActiveMQTextMessageContentRaceTest {
final AtomicReference<String> 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 class ActiveMQTextMessageContentRaceTest {
Thread readerThread = new Thread(() -> {
try {
barrier.await();
- msg.getText();
+ decodeBody(msg);
} catch (Exception e) {
// ignore
}
@@ -145,7 +201,7 @@ public class ActiveMQTextMessageContentRaceTest {
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 class ActiveMQTextMessageContentRaceTest {
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 class ActiveMQTextMessageContentRaceTest {
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);
@@ -191,7 +247,7 @@ public class ActiveMQTextMessageContentRaceTest {
Thread readerThread = new Thread(() -> {
try {
barrier.await();
- msg.getText();
+ decodeBody(msg);
} catch (Exception e) {
// ignore
}
@@ -223,9 +279,9 @@ public class ActiveMQTextMessageContentRaceTest {
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 class ActiveMQTextMessageContentRaceTest {
*/
@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 class ActiveMQTextMessageContentRaceTest {
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;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
