You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cs...@apache.org on 2026/04/02 16:44:39 UTC
(activemq) branch activemq-5.19.x updated: Ensure AMQP protocol marshals messages before passing to broker (#1859) (#1861)
This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch activemq-5.19.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-5.19.x by this push:
new a7da1f8b47 Ensure AMQP protocol marshals messages before passing to broker (#1859) (#1861)
a7da1f8b47 is described below
commit a7da1f8b47c519fbc4bf9b070e3b8faadc4e037c
Author: Christopher L. Shannon <cs...@apache.org>
AuthorDate: Thu Apr 2 12:44:33 2026 -0400
Ensure AMQP protocol marshals messages before passing to broker (#1859) (#1861)
This change brings the behaior inline with other protocols and will
prevent a future race condition during copy/dispatch as the data will
already be marshaled.
This closes #1858
(cherry picked from commit e8f10dd24a26a04892fe9d7b10fd697a2e138d19)
---
.../transport/amqp/protocol/AmqpReceiver.java | 5 ++++
.../activemq/transport/amqp/JMSClientTest.java | 27 ++++++++++++++++++++++
2 files changed, 32 insertions(+)
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpReceiver.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpReceiver.java
index a7ea354d56..13604e4ca9 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpReceiver.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpReceiver.java
@@ -205,6 +205,11 @@ public class AmqpReceiver extends AmqpAbstractReceiver {
}
message.onSend();
+ // GH-1851 - Ensure we marshal the message before passing to the broker
+ // This prevents a race condition later if the message is copyied/marshaled
+ // at this same time. This behavior is in line with how messages are received
+ // using OpenWire (when not using VM) and Stomp.
+ message.beforeMarshall(null);
sendsInFlight++;
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
index c240398bc4..476a2c04e9 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
@@ -26,6 +26,7 @@ import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -52,6 +53,10 @@ import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
+import org.apache.activemq.broker.Broker;
+import org.apache.activemq.broker.BrokerFilter;
+import org.apache.activemq.broker.BrokerPlugin;
+import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.broker.jmx.BrokerView;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.ConnectorViewMBean;
@@ -1358,4 +1363,26 @@ public class JMSClientTest extends JMSClientTestSupport {
assertTrue("Did not receive all messages: " + MSG_COUNT, done.await(15, TimeUnit.SECONDS));
}
+
+ @Override
+ protected void addAdditionalPlugins(List<BrokerPlugin> plugins) throws Exception {
+ super.addAdditionalPlugins(plugins);
+ plugins.add(new BrokerPlugin() {
+
+ @Override
+ public Broker installPlugin(Broker broker) {
+ return new BrokerFilter(broker) {
+
+ @Override
+ public void send(ProducerBrokerExchange producerExchange,
+ org.apache.activemq.command.Message messageSend) throws Exception {
+ super.send(producerExchange, messageSend);
+
+ // The message should always be passed to the broker in a marshaled state
+ assertTrue(messageSend.isMarshalled());
+ }
+ };
+ }
+ });
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact