This is an automated email from the ASF dual-hosted git repository.
robbie pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new 4785995f58 ARTEMIS-3357 Properly compare the subscription address on
client re-attach
4785995f58 is described below
commit 4785995f58876ee719fe2e93780f0d164bc87a68
Author: Timothy Bish <[email protected]>
AuthorDate: Fri Jan 27 16:28:02 2023 -0500
ARTEMIS-3357 Properly compare the subscription address on client re-attach
If the client is using address prefixes to define the routing type along
with
durable subscriptions then on re-attach the compairon to check if the
subscription
address has changed needs to remove the prefix when comparing against the
address
since the prefix isn't propagated when creating the address and will always
fail
resulting in the subscription queue being deleted in error.
---
.../protocol/amqp/broker/AMQPSessionCallback.java | 4 +
.../amqp/proton/ProtonServerSenderContext.java | 12 +-
...leReceiverReconnectWithMulticastPrefixTest.java | 210 +++++++++++++++++++++
3 files changed, 222 insertions(+), 4 deletions(-)
diff --git
a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
index 0603688a96..f28845b914 100644
---
a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
+++
b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
@@ -356,6 +356,10 @@ public class AMQPSessionCallback implements
SessionCallback {
return addressQueryResult;
}
+ public SimpleString removePrefix(SimpleString address) {
+ return serverSession.removePrefix(address);
+ }
+
public void closeSender(final Object brokerConsumer) throws Exception {
final ServerConsumer consumer = ((ServerConsumer) brokerConsumer);
consumer.close(false);
diff --git
a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
index 40af705c38..1068e5c044 100644
---
a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
+++
b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
@@ -1173,11 +1173,15 @@ public class ProtonServerSenderContext extends
ProtonInitializable implements Pr
QueueQueryResult result = sessionSPI.queueQuery(queue,
routingTypeToUse, false);
if (result.isExists()) {
/*
- * If a client reattaches to a durable subscription with
a different filter or address then we must
- * recreate the queue (JMS semantics). However, if the
corresponding queue is managed via the
- * configuration then we don't want to change it
+ * If a client attaches to an existing durable
subscription with a different filter or address then
+ * we must recreate the queue (JMS semantics). However,
if the corresponding queue is managed via the
+ * configuration then we don't want to change it. We must
account for optional address prefixes that
+ * are not carried over into the actual created address
by stripping any prefix value that matches
+ * those configured on the acceptor.
*/
- if (!result.isConfigurationManaged() &&
(!Objects.equals(result.getAddress(), addressToUse) ||
!Objects.equals(result.getFilterString(), simpleStringSelector))) {
+ if (!result.isConfigurationManaged() &&
+ (!Objects.equals(result.getAddress(),
sessionSPI.removePrefix(addressToUse)) ||
+ !Objects.equals(result.getFilterString(),
simpleStringSelector))) {
if (result.getConsumerCount() == 0) {
sessionSPI.deleteQueue(queue);
diff --git
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDurableReceiverReconnectWithMulticastPrefixTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDurableReceiverReconnectWithMulticastPrefixTest.java
new file mode 100644
index 0000000000..2f0355850c
--- /dev/null
+++
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDurableReceiverReconnectWithMulticastPrefixTest.java
@@ -0,0 +1,210 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.amqp;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.postoffice.Binding;
+import org.apache.activemq.artemis.core.postoffice.impl.LocalQueueBinding;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.AddressQueryResult;
+import org.apache.activemq.artemis.core.server.JournalType;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpReceiver;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(Parameterized.class)
+public class AmqpDurableReceiverReconnectWithMulticastPrefixTest extends
JMSClientTestSupport {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Parameterized.Parameters(name = "routingType={0}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][] {
+ {RoutingType.ANYCAST}, {RoutingType.MULTICAST}
+ });
+ }
+
+ @Parameterized.Parameter(0)
+ public RoutingType routingType;
+
+ @Override
+ protected String getConfiguredProtocols() {
+ return "AMQP";
+ }
+
+ @Override
+ protected void createAddressAndQueues(ActiveMQServer server) throws
Exception {
+ // Don't create anything by default since our test relies on prefixes to
define routing types.
+ }
+
+ @Override
+ protected void configureAMQPAcceptorParameters(Map<String, Object> params) {
+ params.put("anycastPrefix", "anycast://");
+ params.put("multicastPrefix", "multicast://");
+ }
+
+ @Override
+ protected void configureAddressPolicy(ActiveMQServer server) {
+ Configuration serverConfig = server.getConfiguration();
+ serverConfig.setJournalType(JournalType.NIO);
+ Map<String, AddressSettings> map = serverConfig.getAddressSettings();
+ if (map.size() == 0) {
+ AddressSettings as = new AddressSettings();
+ map.put("#", as);
+ }
+ Map.Entry<String, AddressSettings> entry =
map.entrySet().iterator().next();
+ AddressSettings settings = entry.getValue();
+ settings.setAutoCreateQueues(true);
+ settings.setDefaultAddressRoutingType(routingType);
+ logger.info("server config, isauto? {}",
entry.getValue().isAutoCreateQueues());
+ logger.info("server config, default address routing type? {}",
entry.getValue().getDefaultAddressRoutingType());
+ }
+
+ @Test(timeout = 60000)
+ public void testReattachToDurableNodeAndTryAndReceiveNewlySentMessage()
throws Exception {
+ final String addressName = "test-address";
+ final String prefixedName = "multicast://" + addressName;
+
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.createConnection());
+ connection.setContainerId(getContainerID());
+ connection.connect();
+
+ final AmqpSession session = connection.createSession();
+
+ AmqpReceiver receiver = session.createDurableReceiver(prefixedName,
getSubscriptionName());
+
+ receiver.detach();
+
+ AddressQueryResult address = getProxyToAddress(addressName);
+
+ assertNotNull(address);
+ assertEquals(Set.of(RoutingType.MULTICAST), address.getRoutingTypes());
+
+ assertEquals(0, lookupSubscriptionQueue().getMessageCount());
+
+ // Recover without lookup as a non-JMS client might do
+ receiver = session.createDurableReceiver(prefixedName,
getSubscriptionName());
+ receiver.flow(1);
+
+ assertEquals(0, lookupSubscriptionQueue().getMessageCount());
+
+ final AmqpSender sender = session.createSender(addressName);
+ final AmqpMessage message = new AmqpMessage();
+
+ message.setMessageId("msg:1");
+ message.setText("Test-Message");
+
+ sender.send(message);
+
+ assertNotNull(receiver.receive(5, TimeUnit.SECONDS));
+
+ assertEquals(1, lookupSubscriptionQueue().getDeliveringCount());
+
+ sender.close();
+ receiver.close();
+
+ connection.close();
+ }
+
+ @Test(timeout = 60000)
+ public void
testReattachToDurableNodeAndTryAndReceivePreviouslySentMessage() throws
Exception {
+ final String addressName = "test-address";
+ final String prefixedName = "multicast://" + addressName;
+
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.createConnection());
+ connection.setContainerId(getContainerID());
+ connection.connect();
+
+ final AmqpSession session = connection.createSession();
+
+ // Recover without lookup as a non-JMS client might do
+ AmqpReceiver receiver = session.createDurableReceiver(prefixedName,
getSubscriptionName());
+
+ receiver.detach();
+
+ AddressQueryResult address = getProxyToAddress(addressName);
+
+ assertNotNull(address);
+ assertEquals(Set.of(RoutingType.MULTICAST), address.getRoutingTypes());
+
+ assertEquals(0, lookupSubscriptionQueue().getMessageCount());
+
+ final AmqpSender sender = session.createSender(addressName);
+ final AmqpMessage message = new AmqpMessage();
+
+ message.setMessageId("msg:1");
+ message.setText("Test-Message");
+
+ sender.send(message);
+
+ assertEquals(1, lookupSubscriptionQueue().getMessageCount());
+
+ receiver = session.createDurableReceiver(prefixedName,
getSubscriptionName());
+
+ receiver.flow(1);
+ assertNotNull(receiver.receive(5, TimeUnit.SECONDS));
+
+ assertEquals(1, lookupSubscriptionQueue().getDeliveringCount());
+
+ sender.close();
+ receiver.close();
+
+ connection.close();
+ }
+
+ private String getContainerID() {
+ return "myContainerID";
+ }
+
+ private String getSubscriptionName() {
+ return "mySubscription";
+ }
+
+ private Queue lookupSubscriptionQueue() {
+ Binding binding = server.getPostOffice().getBinding(new
SimpleString(getContainerID() + "." + getSubscriptionName()));
+ if (binding != null && binding instanceof LocalQueueBinding) {
+ return ((LocalQueueBinding) binding).getQueue();
+ }
+
+ throw new AssertionError("Should have found an existing queue binding
for the durable subscription");
+ }
+
+ private AddressQueryResult getProxyToAddress(String addressName) throws
Exception {
+ return server.addressQuery(SimpleString.toSimpleString(addressName));
+ }
+}
