[
https://issues.apache.org/jira/browse/ARTEMIS-1542?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16286133#comment-16286133
]
ASF GitHub Bot commented on ARTEMIS-1542:
-----------------------------------------
Github user toddbaert commented on the issue:
https://github.com/apache/activemq-artemis/pull/1702
@clebertsuconic @michaelandrepearce
in ClusterConnectionBridge.beforeForward(Message message, SimpleString
forwardingAddress), we call super.beforeForward(message, null). The is why we
lose the address. This code path is executed when broker1 forwards an AMQP
message to broker2. This is why we will only see this in a cluster, when a
producer sends a message to broker1 and a consumer is listening on broker2.
> AMQPMessage can throw / not check for null
> ------------------------------------------
>
> Key: ARTEMIS-1542
> URL: https://issues.apache.org/jira/browse/ARTEMIS-1542
> Project: ActiveMQ Artemis
> Issue Type: Bug
> Components: AMQP, Broker
> Affects Versions: 2.4.0
> Environment: Linux Ubuntu 16.04 LTS
> ActiveMQ Artemis 2.4.0
> Tested with qpid-python clients from qpid-cpp-1.36.0 package
> Also tested and reproducable with the attached program send-test.cpp (based
> on qpid proton)
> Reporter: AWETTT
> Attachments: send-test.cpp
>
>
> I built active/active cluster with two brokers
> A message was sent to a queue using AMQP 1.0 to broker1. A client at broker2
> tried to read it.
> But broker2 throws an exception and the message is not sent to the client:
> {noformat}
> 09:34:53,146 ERROR [org.apache.activemq.artemis.core.server] AMQ224016:
> Caught exception: java.lang.NullPointerException
> at
> org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage.setAddress(AMQPMessage.java:612)
> [artemis-amqp-protocol-2.4.0.jar:]
> at
> org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage.setAddress(AMQPMessage.java:63)
> [artemis-amqp-protocol-2.4.0.jar:]
> at
> org.apache.activemq.artemis.core.server.impl.ServerSessionImpl.send(ServerSessionImpl.java:1368)
> [artemis-server-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.core.server.impl.ServerSessionImpl.send(ServerSessionImpl.java:1311)
> [artemis-server-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.core.server.impl.ServerSessionImpl.send(ServerSessionImpl.java:1304)
> [artemis-server-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.core.protocol.core.ServerSessionPacketHandler.onSessionSend(ServerSessionPacketHandler.java:690)
> [artemis-server-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.core.protocol.core.ServerSessionPacketHandler.onMessagePacket(ServerSessionPacketHandler.java:290)
> [artemis-server-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.utils.actors.Actor.doTask(Actor.java:33)
> [artemis-commons-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.utils.actors.ProcessorBase$ExecutorTask.run(ProcessorBase.java:53)
> [artemis-commons-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.utils.actors.OrderedExecutor.doTask(OrderedExecutor.java:42)
> [artemis-commons-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.utils.actors.OrderedExecutor.doTask(OrderedExecutor.java:31)
> [artemis-commons-2.4.0.jar:2.4.0]
> at
> org.apache.activemq.artemis.utils.actors.ProcessorBase$ExecutorTask.run(ProcessorBase.java:53)
> [artemis-commons-2.4.0.jar:2.4.0]
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> [rt.jar:1.8.0_151]
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> [rt.jar:1.8.0_151]
> at java.lang.Thread.run(Thread.java:748) [rt.jar:1.8.0_151]
> {noformat}
> *The queue is not created on the fly but configured in the brokers, both
> brokers have this config:*
> {noformat}
> <queues>
> <queue name="awe.test.queue">
> <address>awe.test.queue</address>
> <durable>true</durable>
> </queue>
> </queues>
> {noformat}
> *The cluster configuration is pretty much copied from the examples:*
> {noformat}
> <!-- Clustering configuration -->
> <broadcast-groups>
> <broadcast-group name="my-broadcast-group">
> <group-address>${udp-address:231.7.7.7}</group-address>
> <group-port>9876</group-port>
> <broadcast-period>100</broadcast-period>
> <connector-ref>netty-connector</connector-ref>
> </broadcast-group>
> </broadcast-groups>
> <discovery-groups>
> <discovery-group name="my-discovery-group">
> <group-address>${udp-address:231.7.7.7}</group-address>
> <group-port>9876</group-port>
> <refresh-timeout>10000</refresh-timeout>
> </discovery-group>
> </discovery-groups>
> <cluster-connections>
> <cluster-connection name="my-cluster">
> <connector-ref>netty-connector</connector-ref>
> <retry-interval>500</retry-interval>
> <use-duplicate-detection>true</use-duplicate-detection>
> <message-load-balancing>STRICT</message-load-balancing>
> <max-hops>1</max-hops>
> <discovery-group-ref discovery-group-name="my-discovery-group"/>
> </cluster-connection>
> </cluster-connections>
> <!-- a colocated server that will allow shared store full backups to be
> requested-->
> <ha-policy>
> <shared-store>
> <colocated>
> <backup-port-offset>100</backup-port-offset>
> <backup-request-retries>-1</backup-request-retries>
>
> <backup-request-retry-interval>2000</backup-request-retry-interval>
> <max-backups>1</max-backups>
> <request-backup>true</request-backup>
> <master>
> <failover-on-shutdown>true</failover-on-shutdown>
> </master>
> <slave>
> <failover-on-shutdown>true</failover-on-shutdown>
> </slave>
> </colocated>
> </shared-store>
> </ha-policy>
> {noformat}
> *qpid-send and qpid-receive were used as test clients. Sending is done like:*
> {noformat}
> qpid-send -b localhost:9800 -a awe.test.queue
> '--connection-option={protocol:amqp1.0}' --content-string 'test message Do 7.
> Dez 09:31:51 CET 2017' --durable=yes
> {noformat}
> +Grabbed the message from the logfile. It looks like this:+
> {noformat}
> 0 -> @transfer(20) [handle=0, delivery-id=0,
> delivery-tag=b"\x00\x00\x00\x00", message-format=0, settled=false,
> more=false]
> (79)
> "\x00Sp\xc0\x04\x02AP\x00\x00St\xc1\x14\x04\xa1\x02snR\x01\xa1\x02ts\x81\x14\xfd\xf8\xcb\xdaG\x94U\x00Sw\xa1(test
> message Do 7. Dez 09:31:51 CET 2017"
> {noformat}
> *Reading is done using:*
> {noformat}
> qpid-receive -b localhost:9802 -a awe.test.queue
> '--connection-option={protocol:amqp1.0}' -t --timeout 500
> {noformat}
> I can see, that AMQPMessage.getAddress() can return null:
> {code:java}
> @Override
> public String getAddress() {
> if (address == null) {
> Properties properties = getProtonMessage().getProperties();
> if (properties != null) {
> return properties.getTo();
> } else {
> return null;
> }
> } else {
> return address;
> }
> }
> {code}
> But ServerSessionImpl.send() does not check for null, but it should. This is
> because SimpleString.toSimpleString() can return null if the input is null:
> {code:java}
> SimpleString address = message.getAddressSimpleString();
> if (defaultAddress == null && address != null) {
> defaultAddress = address;
> }
> if (address == null) {
> // We don't want to force a re-encode when the message gets sent to
> the consumer
> message.setAddress(defaultAddress);
> }
> {code}
> SimpleString:
> {code:java}
> public static SimpleString toSimpleString(final String string) {
> if (string == null) {
> return null;
> }
> return new SimpleString(string);
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
