From 47e20300b933d96a2057cb6601c7f0b49d4dfc0d Mon Sep 17 00:00:00 2001 From: "Christopher L. Shannon" Date: Sun, 5 Jul 2026 20:00:56 -0400 Subject: [PATCH 1/3] Fix TextMessage body thread safety 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. --- .attach_pid78822 | 0 .../activemq/command/ActiveMQTextMessage.java | 97 ++++++++++++++----- .../org/apache/activemq/command/Message.java | 3 +- 3 files changed, 77 insertions(+), 23 deletions(-) create mode 100644 .attach_pid78822 diff --git a/.attach_pid78822 b/.attach_pid78822 new file mode 100644 index 00000000000..e69de29bb2d 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 a5df4164f0a..ffbc543d521 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,7 +45,8 @@ 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() { @@ -54,7 +55,7 @@ public Message copy() { return copy; } - private void copy(ActiveMQTextMessage copy) { + private synchronized void copy(ActiveMQTextMessage copy) { super.copy(copy); copy.text = text; } @@ -72,18 +73,29 @@ public String getJMSXMimeType() { @Override public void setText(String text) throws MessageNotWriteableException { checkReadOnlyBody(); - this.text = text; - setContent(null); + synchronized (this) { + this.text = text; + setContent(null); + } } @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); + synchronized (this) { + content = getContent(); + 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 +136,57 @@ public void beforeMarshall(WireFormat wireFormat) throws IOException { @Override public void storeContentAndClear() { - storeContent(); - text=null; + storeContent(true); } @Override public void storeContent() { + storeContent(false); + } + + private void storeContent(final boolean clear) { try { ByteSequence content = getContent(); String text = this.text; + // Both of these are volatile reads, we can just skip + // if state is already correct 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); + synchronized (this) { + // Double-checked locking, re-read under lock + content = getContent(); + text = this.text; + // re-check if we need to store 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(); + this.content = content = bytesOut.toByteSequence(); + } + // We are in a synchornized block so at this point we know we only need + // to clear if content is not null and text is not null + if (clear && text != null && content != null) { + this.text = null; + } + } + } else if (clear && content != null && text != null) { + synchronized (this) { + // re-read under lock + content = getContent(); + text = this.text; + // Double-checked locking + // We are in a synchornized block so at this point we know we only need + // to clear if content is not null and text is not null + if (text != null && content != null) { + this.text = null; + } } - DataOutputStream dataOut = new DataOutputStream(os); - MarshallingSupport.writeUTF8(dataOut, text); - dataOut.close(); - setContent(bytesOut.toByteSequence()); } } catch (IOException e) { throw new RuntimeException(e); @@ -156,11 +198,19 @@ public void storeContent() { @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; + } + } } @Override public boolean isContentMarshalled() { + // volatile reads, should be fine to just check without synchronized return content != null || text == null; } @@ -177,13 +227,16 @@ public boolean isContentMarshalled() { */ @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; + ByteSequence content = getContent(); if (size == 0 && content == null && text != null) { size = getMinimumMessageSize(); if (marshalledProperties != 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 06e0aebeea6..5f48c78d5da 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,7 +81,7 @@ 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; @@ -715,6 +715,7 @@ public int decrementReferenceCount() { @Override public int getSize() { int minimumMessageSize = getMinimumMessageSize(); + ByteSequence content = this.content; if (size < minimumMessageSize || size == 0) { size = minimumMessageSize; if (marshalledProperties != null) { From 48746649a470d1dc108e003e432aaa3f099a67ed Mon Sep 17 00:00:00 2001 From: "Christopher L. Shannon" Date: Mon, 6 Jul 2026 10:10:30 -0400 Subject: [PATCH 2/3] Add sync to setContent() --- .../org/apache/activemq/command/ActiveMQTextMessage.java | 7 +++++++ 1 file changed, 7 insertions(+) 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 ffbc543d521..370835863dd 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 @@ -79,6 +79,13 @@ public void setText(String text) throws MessageNotWriteableException { } } + // 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(); From 20780ce45b9a220ba2922f3d1aa2c056a89247bd Mon Sep 17 00:00:00 2001 From: "Christopher L. Shannon" Date: Mon, 6 Jul 2026 13:14:43 -0400 Subject: [PATCH 3/3] rework fix to prevent race condition --- .../activemq/command/ActiveMQTextMessage.java | 90 ++++++++----------- .../org/apache/activemq/command/Message.java | 5 +- 2 files changed, 41 insertions(+), 54 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 370835863dd..0ace802c8ce 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 @@ -51,15 +51,13 @@ public class ActiveMQTextMessage extends ActiveMQMessage implements TextMessage @Override public Message copy() { ActiveMQTextMessage copy = new ActiveMQTextMessage(); - copy(copy); + synchronized (this) { + super.copy(copy); + copy.text = text; + } return copy; } - private synchronized void copy(ActiveMQTextMessage copy) { - super.copy(copy); - copy.text = text; - } - @Override public byte getDataStructureType() { return DATA_STRUCTURE_TYPE; @@ -88,15 +86,12 @@ public synchronized void setContent(ByteSequence content) { @Override public String getText() throws JMSException { - ByteSequence content = getContent(); String text = this.text; - if (text == null && content != null) { + if (text == null) { synchronized (this) { - content = getContent(); text = this.text; - // Double-checked locking, re-check under lock if we need - // to decode + // Double-checked locking, re-check under lock if we need to decode if (text == null && content != null) { this.text = text = decodeContent(content); setContent(null); @@ -104,6 +99,7 @@ public String getText() throws JMSException { } } } + return text; } @@ -143,26 +139,22 @@ public void beforeMarshall(WireFormat wireFormat) throws IOException { @Override public void storeContentAndClear() { - storeContent(true); + // 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() { - storeContent(false); - } - - private void storeContent(final boolean clear) { try { - ByteSequence content = getContent(); - String text = this.text; - // Both of these are volatile reads, we can just skip - // if state is already correct - if (content == null && text != null) { + // Content is volatile so if it's not null we can skip and do nothing + if (content == null) { synchronized (this) { - // Double-checked locking, re-read under lock - content = getContent(); - text = this.text; - // re-check if we need to store under lock + // Double-checked locking, re-check state under lock if (content == null && text != null) { ByteArrayOutputStream bytesOut = new ByteArrayOutputStream(); OutputStream os = bytesOut; @@ -174,24 +166,7 @@ private void storeContent(final boolean clear) { DataOutputStream dataOut = new DataOutputStream(os); MarshallingSupport.writeUTF8(dataOut, text); dataOut.close(); - this.content = content = bytesOut.toByteSequence(); - } - // We are in a synchornized block so at this point we know we only need - // to clear if content is not null and text is not null - if (clear && text != null && content != null) { - this.text = null; - } - } - } else if (clear && content != null && text != null) { - synchronized (this) { - // re-read under lock - content = getContent(); - text = this.text; - // Double-checked locking - // We are in a synchornized block so at this point we know we only need - // to clear if content is not null and text is not null - if (text != null && content != null) { - this.text = null; + setContent(bytesOut.toByteSequence()); } } } @@ -215,9 +190,9 @@ public void clearUnMarshalledState() throws JMSException { } } + // We need to sync because both variables need to be read independently @Override - public boolean isContentMarshalled() { - // volatile reads, should be fine to just check without synchronized + public synchronized boolean isContentMarshalled() { return content != null || text == null; } @@ -242,14 +217,23 @@ public void clearBody() throws JMSException { @Override public int getSize() { - String text = this.text; - ByteSequence content = getContent(); - 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(); } @@ -258,7 +242,7 @@ public int getSize() { 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 5f48c78d5da..11ccd306fa4 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 @@ -86,7 +86,7 @@ public abstract class Message extends BaseCommand implements MarshallAware, Mess protected DataStructure dataStructure; protected int redeliveryCounter; - protected int size; + protected volatile int size; protected Map properties; protected boolean readOnlyProperties; protected boolean readOnlyBody; @@ -716,14 +716,17 @@ public int decrementReferenceCount() { 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; }