Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Timothy Bish 2023-01-27 16:28:02 -05:00
parent 009f539406
commit 4785995f58
3 changed files with 222 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
View File

@ -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);

12
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
View File

@ -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);

210
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDurableReceiverReconnectWithMulticastPrefixTest.java Normal file
View File

@ -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));
}
}