AMQ-9855: VMTransport, Defensive copy of messages to prevent mutation

[pradeep85841]

Updated VMTransport.java doDispatch() to create a defensive copy of ActiveMQMessage to prevent shared message mutation.

Verified locally with a helper: original message body remains unchanged and copy is correctly dispatched.

No new files added; only VMTransport.java modified.

[mattrpav]

Have you observed a bug or problem with message mutation? If so, please provide the scenario and reproducible test case.

ActiveMQConnectionFactory already has copyMessageOnSend enabled by default.

I need help understanding why (in effect) 2 copies are needed.

[jbonofre]

@pradeep85841 can you please provide some details here ? Do you have an issue/test case ? It's not obvious to me if it's cosmetic or actual issue.

[pradeep85841]

This is based on issue with vm:// and topics (AMQ-9855).

With VMTransport the same ActiveMQMessage instance is dispatched to multiple consumers. If one consumer reads or mutates the body (Camel split/processor does this), other consumers can see an empty body. This does not happen over tcp:// because marshal/unmarshal creates a copy.

copyMessageOnSend only applies at the producer and broker boundary. The problem here happens inside the broker during dispatch, so that setting does not help.

The change makes VM transport behave consistently with TCP transport and avoids shared mutable state

Happy to add a test if needed.

[jbonofre]

OK, let me take a new look. I'm adding the Jira id in the title of the PR to avoid confusion. Thanks.

[mattrpav]

This probably needs a config flag that is a corollary to the copyMessageOnSend

ie. copyMessageOnDispatch

[pradeep85841]

Sounds good, thanks.
Happy to make this configurable and I’ll wait for next steps.

[pradeep85841]

Applied a defensive copy in VMTransport.java to ensure each consumer on vm:// topics receives an independent message.

Added VMTransportDefensiveCopyTest.java to reproduce the issue and verify the fix.

Confirms that message bodies are not shared/mutated between consumers.

Addresses the intermittent null/empty body problem reported in AMQ-9855.

[mattrpav]

Should this be enabled on the client connection factory? JMS messages are supposed to be immutable. If someone is modifying it post recv, seems like that is on the app-side to ensure the spec is not violated when using vm transport.

[pradeep85841]

Thanks for the suggestion.

The issue in AMQ-9855 happens inside the vm:// transport dispatch path. The same ActiveMQMessage instance is delivered to multiple consumers in-memory. With tcp://, marshal/unmarshal creates separate instances, so this does not occur.

So this is not about client code mutating a received JMS message, it’s about vm:// sharing the same object reference across consumers, which makes its behaviour inconsistent with other transports.

[mattrpav]

The consumer-side may perform the copy, no?

[pradeep85841]

The consumer side could defensively copy the message, but that would not solve the root cause. By the time the consumer performs the copy, the same message instance may already have been observed or mutated by another consumer. In that case, the copy would simply duplicate an already-modified state.

[jbonofre]

This is an interesting PR because it's a good way to have the VM transport behavior close to another "remote" transport connector. However, there's a purpose about the VM: direct communication without marshaling.
I will take a deeper look, but very interesting, thanks @pradeep85841 !

[jbonofre]

@mattrpav I would appreciate that you don't remove me from the reviewer when I requested it. Thanks.

[jeanouii]

@pradeep85841 That's a very good catch. And I think it makes a lot of sense to have this to make sure vm transport behaves the same as any remote protocol.
In EJB world, we have also @Local and @Remote interfaces. Some app servers, like Apache TomEE have the same kind of optimization. If you call a @Remote interface within the same JVM, the server will mimic the serialization / deserialization before actually calling the target bean in the JVM.
That makes a lot of sense.

[pradeep85841]

Thanks for the feedback.

I’ve removed the Dockerfile and entrypoint.sh changes from this PR since they belong to the separate non-root Docker work. This PR is now strictly focused on the VM transport change for AMQ-9855.

[jeanouii]

Overall looking good to me. Great job!

See comments bellow

[jeanouii]

This is discouraged in the project, even though it's not a blocker

[jeanouii]

Same here FYI

[jeanouii]

I'm probably blind or my search/replace does not work properly. Where are the 2 fields used so far?

[jeanouii]

I'm wondering if it's desire or not to be honest. The goal to me of this PR (and it's great) is to have VM transport behave the same as other remote transport. Benefit being that others in the same JVM can't mutate the message. Great!

Now if we can't serialize/de-serialize to create a deep copy and we still send the original, we might introduce a case where VM does work when remote does not. So I'm tempted to just fail here. What do you think?

[mattrpav]

command.copy() should suffice here

edit: Agree, we definitely don't want to instantiate an OpenWireFormat object per-message here.

[jeanouii]

Same

[jeanouii]

Should we have a return after this one to avoid the second dispatch bellow?

[jeanouii]

This PR should be rebased against apache/main to avoid this class to be here

[cshannon]

In the past I have also noticed some odd behavior with the VM transport when using clients (producers/consumers) that could be a thread safety bug (it is used for bridges though and seems fine there so that is interesting).

However, this proposed fix in this PR doesn't make sense to me at all because the entire point is to NOT need to marshal the message. If you want to marshal the object then you should just use the TCP transport as you lose the benefits.

The VM transport is supposed to already do a deep copy during dispatch to prevent issues with multi-threading. It's possible there is an issue with that, so that's probably where to look for a fix vs adding in marshaling.

Here is where the copying is doing on dispatch in the connection:

activemq/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java

Lines 1906 to 1929 in a8fa4b0

	public Response processMessageDispatch(MessageDispatch md) throws Exception {
	waitForTransportInterruptionProcessingToComplete();
	ActiveMQDispatcher dispatcher = dispatchers.get(md.getConsumerId());
	if (dispatcher != null) {
	// Copy in case a embedded broker is dispatching via
	// vm://
	// md.getMessage() == null to signal end of queue
	// browse.
	Message msg = md.getMessage();
	if (msg != null) {
	msg = msg.copy();
	msg.setReadOnlyBody(true);
	msg.setReadOnlyProperties(true);
	msg.setRedeliveryCounter(md.getRedeliveryCounter());
	msg.setConnection(ActiveMQConnection.this);
	msg.setMemoryUsage(null);
	md.setMessage(msg);
	}
	dispatcher.dispatch(md);
	} else {
	LOG.debug("{} no dispatcher for {} in {}", this, md, dispatchers);
	}
	return null;
	}

I would start there as to investigating if that is the issue

[jbonofre]

I think @cshannon has a very good point. The VM purpose is to "bypass" the marshaling (direct JVM communication). It was my first question about the use case. If the issue is about multi-threading/consumer, it's worth to investigate there (and not change the intentional behavior).

[jeanouii]

In the past I have also noticed some odd behavior with the VM transport when using clients (producers/consumers) that could be a thread safety bug (it is used for bridges though and seems fine there so that is interesting).

However, this proposed fix in this PR doesn't make sense to me at all because the entire point is to NOT need to marshal the message. If you want to marshal the object then you should just use the TCP transport as you lose the benefits.

The VM transport is supposed to already do a deep copy during dispatch to prevent issues with multi-threading. It's possible there is an issue with that, so that's probably where to look for a fix vs adding in marshaling.

Here is where the copying is doing on dispatch in the connection:

activemq/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java

Lines 1906 to 1929 in a8fa4b0

	public Response processMessageDispatch(MessageDispatch md) throws Exception {
	waitForTransportInterruptionProcessingToComplete();
	ActiveMQDispatcher dispatcher = dispatchers.get(md.getConsumerId());
	if (dispatcher != null) {
	// Copy in case a embedded broker is dispatching via
	// vm://
	// md.getMessage() == null to signal end of queue
	// browse.
	Message msg = md.getMessage();
	if (msg != null) {
	msg = msg.copy();
	msg.setReadOnlyBody(true);
	msg.setReadOnlyProperties(true);
	msg.setRedeliveryCounter(md.getRedeliveryCounter());
	msg.setConnection(ActiveMQConnection.this);
	msg.setMemoryUsage(null);
	md.setMessage(msg);
	}
	dispatcher.dispatch(md);
	} else {
	LOG.debug("{} no dispatcher for {} in {}", this, md, dispatchers);
	}
	return null;
	}

I would start there as to investigating if that is the issue

Thanks for the background @cshannon

[pradeep85841]

I’ve applied all requested changes: cleaned up imports and unused fields in VMTransport.java, ensured proper deep-copy handling in oneway() so VM transport behaves like other remote transports, added a return to avoid duplicate dispatch, and updated the unit tests to focus on the AMQ-9855 scenario. All changes are tested and ready for review.

[cshannon]

It looks like a bunch of binary files were committed to this

[pradeep85841]

I’ve cleaned up the branch and rebased against main. The PR now only contains the intended changes:

VMTransport.java (defensive copy in oneway() for AMQ-9855)

The corresponding unit test

[cshannon]

The latest changes here still do not address this comment I made: #1659 (review)

This is still marshaling which we should not be doing, with the VM transport we should just be using the message copy methods.

BUT, the question is still what is the real issue and why is this happening? As pointed out, we already copy the message on dispatch to consumers.

The test you wrote does fail without your changes but ONLY because you are only comparing the String instances to see if they are the same, which they won't be as you are using OpenWire. However, with calling copy() the String instances are the same as Strings are immutable. If you modified your test to compare instance equality for the TextMessage instances and not just the bodies the test would pass because we copy on dispatch.

My guess is the real issue could be a race condition during the copy or when the body is converted between states. The messages themselves are generally not thread safe which is why we copy on dispatch so each consumer gets their own copy, but some messages can toggle state..they can convert transparently between marshaled body and the in memory usable body (same with message properties).

@pradeep85841 - Are you only using Text messages?

So for a text message, the message will either store the body as a String or it will store it as a buffer and can switch. My guess is things are breaking during that switch as multiple threads are probably calling some of these methods concurrently:

activemq/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java

Lines 73 to 181 in 649aa0a

	public void setText(String text) throws MessageNotWriteableException {
	checkReadOnlyBody();
	this.text = text;
	setContent(null);
	}
	@Override
	public String getText() throws JMSException {
	ByteSequence content = getContent();
	if (text == null && content != null) {
	text = decodeContent(content);
	setContent(null);
	setCompressed(false);
	}
	return text;
	}
	private String decodeContent(ByteSequence bodyAsBytes) throws JMSException {
	String text = null;
	if (bodyAsBytes != null) {
	InputStream is = null;
	try {
	is = new ByteArrayInputStream(bodyAsBytes);
	if (isCompressed()) {
	is = new InflaterInputStream(is);
	}
	DataInputStream dataIn = new DataInputStream(is);
	text = MarshallingSupport.readUTF8(dataIn);
	dataIn.close();
	} catch (IOException ioe) {
	throw JMSExceptionSupport.create(ioe);
	} finally {
	if (is != null) {
	try {
	is.close();
	} catch (IOException e) {
	// ignore
	}
	}
	}
	}
	return text;
	}
	@Override
	public void beforeMarshall(WireFormat wireFormat) throws IOException {
	super.beforeMarshall(wireFormat);
	storeContentAndClear();
	}
	@Override
	public void storeContentAndClear() {
	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);
	}
	DataOutputStream dataOut = new DataOutputStream(os);
	MarshallingSupport.writeUTF8(dataOut, text);
	dataOut.close();
	setContent(bytesOut.toByteSequence());
	}
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	}
	// 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();
	this.text = null;
	}
	@Override
	public boolean isContentMarshalled() {
	return content != null || text == null;
	}
	/**
	* Clears out the message body. Clearing a message's body does not clear its
	* header values or property entries. <p/>
	* <P>
	* If this message body was read-only, calling this method leaves the
	* message body in the same state as an empty body in a newly created
	* message.
	*
	* @throws JMSException if the JMS provider fails to clear the message body
	* due to some internal error.
	*/
	@Override
	public void clearBody() throws JMSException {
	super.clearBody();
	this.text = null;
	}

This might apply to other message types as well but this was the most noticeable type because it tries to only store either the String or byte buffer to save space.

Going back a long time ago, I actually made some changes here to try and prevent issues without adding synchronization but it's not perfect: https://issues.apache.org/jira/browse/AMQ-5857

So we just need to figure out the exact issue, if it's the copy() that is being done on dispatch (maybe multiple connections are calling copy at the same time on dispatch) we may just need to add some sort of synchronization there. But we need ot identify the real cause, a good first step might be to just test what happens if you apply the "synchronized" key word on the copy() method or also on the other methods that due mutations like getText(), setText(), storeContent() etc and re-run things to see if the issue is gone to help narrow it down to that.

[pradeep85841]

@cshannon, Thank you for the insight. That makes perfect sense—the issue likely isn't the lack of a copy, but a race condition during the lazy state-toggle within ActiveMQTextMessage when copy() or getText() is called concurrently.

I am moving away from the marshaling approach. I'll test adding synchronization to the state-changing methods in ActiveMQTextMessage to see if that stabilizes the copy() process during VM dispatch. I'll update the PR once I verify this identifies the root cause

[pradeep85841]

I have refactored the fix to address the root cause at the command level rather than the transport level, as suggested.

The Issue: Under high concurrency, ActiveMQTextMessage suffered from a race condition where the internal state "toggle" (moving between the text field and the ByteSequence content) was not atomic. This led to a "double null" state where one thread cleared the unmarshalled text before the other had finished reading the raw content.

The Fix:

Thread Safety: Added synchronized to getText(), setText(), and state-toggling methods in ActiveMQTextMessage.

State Preservation: Updated clearUnMarshalledState() to ensure that the marshalled content is stored before the unmarshalled state is cleared.

Performance: This approach preserves the efficiency of the VM Transport by avoiding a full transport-level marshal, while ensuring data integrity.

Validation: Added ActiveMQTextMessageStressTest.java, which reproduces the null body issue and confirms the fix under high thread contention.

[jeanouii]

Looks good.
Synchronized is good enough and it's a perf killer only on vm transport. Other transports would marshall/unmarshall and would use a different instance. On vm, it creates contentions but that's way better than messing up with the data. Using RentrantReadWriteLock would not improve the things because all methods basically mutate under the cover.

[jeanouii]

The synchronized bellow on this is accurate because the other synchronized methods are also on this. All good
Wondering why the internal copy(ActiveMQTexteMessage) moved from private to protected?
I don't see subclasses or I missed it, so no reason to expose our internals

[jeanouii]

It's a minor detail and question, but Chris would provide a way better feedback for you :-)

[pradeep85841]

@jeanouii That's a fair point regarding the visibility. I initially changed it to protected to follow the pattern of the parent ActiveMQMessage copy logic, but I agree that since there are no current subclasses of ActiveMQTextMessage, it's better to keep it encapsulated.

I'm going to wait for @cshannon's final feedback before pushing any more updates. This way, I can consolidate all requested changes—including reverting this visibility to private—into a single final push to keep the history clean. Thanks for catching that!

[cshannon]

I agree, leaving it private here makes sense unless we need to change it (we can always change it later)

[cshannon]

Looks good. Synchronized is good enough and it's a perf killer only on vm transport. Other transports would marshall/unmarshall and would use a different instance. On vm, it creates contentions but that's way better than messing up with the data. Using RentrantReadWriteLock would not improve the things because all methods basically mutate under the cover.

Agreed, synchronized here is fine to use and as you pointed out we don't gain anything from a ReadWrite lock as we need a write lock for basically all methods. I wasn't around for the original implementation of classes but it looked like the goal was to avoid synchronization as messages were meant to be copied if used across threads. Unfortunately that pattern breaks with the VM transport as we see due to the nature of dispatching to multiple consumers. I don't think we see this issue with network bridges because there is only one consumer for the VM transports in bridges (the broker itself).

The performance impact for single threaded applications using synchronized that have no contention is pretty minimal (there's technically some hit but unlikely to be noticeable) and even with contention as we can see correctness is more important. I'd rather the application take a couple extra ms to dispatch all the copies of the message than to receive a null body.

I was looking more at AMQ-5857 and the discussion to try and remember why we didn't add synchronized back then when trying to clear memory to avoid storing the data twice and OOM errors and i assume we just didn't have good enough tests to show the real issue. Plus, it was like 10 years ago so I was new to ActiveMQ :)

I haven't had time to fully review this yet but will either tomorrow or Monday and will post more feedback.

[cshannon]

Overall this looks pretty good, I made a few comments in line. The test nicely demonstrates the race condition issue and verifies the fix so good work with that.

After reviewing this I have a couple of thoughts:

1. I am thinking we probably need to add synchronized key words to the other methods in ActiveMQTextMessage. I think clearBody() probably needs it, but I'm also wondering about isContentMarshalled(). The other option is to mark the variables as volatile but if we are using synchronized we don't need to do that of course. I'm not entirely sure we need to mark every method as the goal here is to copy the messages and eventually each thread (consumer, etc) would have their own copy so at that point the synchronization isn't a big deal as long as during the copy we are protected. But it's probably best to play it safe and go all the way and finish marking those other methods synchronized or use volatile so the changes are visible.
2. The other thought I had was the change here should really be applied to other message types as well, if relevant. For example, Map messages will have the same issue. So for consistency I think we need to fix all the messages. We could either fix them as part of this issue or crate a follow on Jira/Issue to fix the others. I don't really care as long as we track it and don't forget. I looked just at Map message but the other types need to be analyzed closer to see what if anything needs fixing.

[cshannon]

DURABLE_CONSUMERS is always 0, so this should either be changed or removed.

[cshannon]

Nitpicking but you should use braces even for one line statements.

[cshannon]

I wouldn't use print stacktrace here, the exception should be handled better. We should either fail the test or in this case maybe just log it at debug level if we don't care.

[cshannon]

Indentation here is off and needs to be fixed

[cshannon]

Indentation here is off and needs to be fixed

[cshannon]

I agree, leaving it private here makes sense unless we need to change it (we can always change it later)

[cshannon]

@pradeep85841 - One more thing, did you re-run your original setup with the new fixes here? Did you verify that the synchronization actually fixes your issue in a real data flow? The test of course tries to simulate the issue but we are still not exactly sure where the race is happening (but of course thing it has to do with copying or something on dispatch). Anyways it would just be good to verify your original issue is indeed fixed by this (I think it will be)

[cshannon]

The only other thing that I keep thinking is if we are going to go ahead and add synchronization here it would be nice to better understand why the issue happens at all. Just thinking about it more there really shouldn't be anything else touching the message when it is copied during dispatch and then each consumer should have its own copy so it makes me wonder if there's something else weird about camel or something that is triggering multiple threads to touch the message at the same time. Clients also have the ability to disable copying during send which could be an issue if set.

Mostly I just want to make sure that we are solving an issue with the actual broker and not a weird setup that is causing it.

[pradeep85841]

I have updated to address the feedback from @cshannon and @jeanouii. This revision ensures total thread safety for the VM Transport "defensive copy" logic while adhering to the project's coding standards and memory-management goals.

Key Changes:

Full Synchronization: Per the suggestion to "go all the way," I have applied the synchronized keyword to all state-dependent methods in ActiveMQTextMessage. This includes getText(), setText(), getSize(), clearBody(), and clearUnMarshalledState().

Data Integrity: In clearUnMarshalledState(), I ensured that storeContent() is called (if bytes are missing) before the local text reference is nulled. This prevents the "double-null" scenario where a message could lose its content entirely during a race.

Encapsulation: Reverted the copy(ActiveMQTextMessage) helper method to private visibility.

Stress Test Improvements: * Replaced printStackTrace with SLF4J logging.

Added proper braces {} to all conditional blocks.

Refined the concurrency logic using an ExecutorService and Future validation to simulate high-pressure dispatching across multiple durable and non-durable subscribers.

Optimized the test payload and iteration count to verify the fix under high contention without triggering OutOfMemoryError on restricted heap environments.

Verification: The updated ActiveMQTextMessageStressTest confirms that even when multiple consumers hammer getText() and clearUnMarshalledState() concurrently on messages dispatched via the VM transport, the data remains intact and instances remain independent.

The fix has been verified in original environment and the test suite is now passing with a clean exit.