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 b131c3d266 Fix TextMessage body thread safety (#2188)
b131c3d266 is described below
commit b131c3d2661ad28236a5ac18826e7d88cd3cb976
Author: Christopher L. Shannon <[email protected]>
AuthorDate: Thu Jul 9 09:05:47 2026 -0400
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.
---
.../activemq/command/ActiveMQTextMessage.java | 124 ++++++++++++++-------
.../java/org/apache/activemq/command/Message.java | 8 +-
2 files changed, 90 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;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
