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 6186805b69 ARTEMIS-5090 MirrorSNF is not supposed to expiry on messages
6186805b69 is described below
commit 6186805b6969318a94a3a7ad418abfad4444804b
Author: Clebert Suconic <[email protected]>
AuthorDate: Tue Oct 8 09:33:57 2024 -0400
ARTEMIS-5090 MirrorSNF is not supposed to expiry on messages
---
.../amqp/broker/ProtonProtocolManager.java | 3 +-
.../artemis/core/server/impl/QueueImpl.java | 17 ++-
.../amqp/connect/AMQPMirrorExpiryQueueTest.java | 124 +++++++++++++++++++++
3 files changed, 138 insertions(+), 6 deletions(-)
diff --git
a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java
b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java
index 147a8a6de9..2c9c8a1679 100644
---
a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java
+++
b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java
@@ -31,6 +31,7 @@ import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
import
org.apache.activemq.artemis.core.remoting.impl.netty.NettyServerConnection;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.impl.QueueImpl;
import org.apache.activemq.artemis.core.server.management.Notification;
import org.apache.activemq.artemis.core.server.management.NotificationListener;
import
org.apache.activemq.artemis.protocol.amqp.client.ProtonClientProtocolManager;
@@ -64,7 +65,7 @@ public class ProtonProtocolManager extends
AbstractProtocolManager<AMQPMessage,
private static final List<String> websocketRegistryNames =
Arrays.asList("amqp");
- public static final String MIRROR_ADDRESS = "$ACTIVEMQ_ARTEMIS_MIRROR";
+ public static final String MIRROR_ADDRESS = QueueImpl.MIRROR_ADDRESS;
private final List<AmqpInterceptor> incomingInterceptors = new
ArrayList<>();
private final List<AmqpInterceptor> outgoingInterceptors = new
ArrayList<>();
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index d86a512a31..46b472510c 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -142,6 +142,9 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
protected static final int CRITICAL_CONSUMER = 3;
protected static final int CRITICAL_CHECK_DEPAGE = 4;
+ // The prefix for Mirror SNF Queues
+ public static final String MIRROR_ADDRESS = "$ACTIVEMQ_ARTEMIS_MIRROR";
+
private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private static final AtomicIntegerFieldUpdater<QueueImpl>
dispatchingUpdater = AtomicIntegerFieldUpdater.newUpdater(QueueImpl.class,
"dispatching");
private static final AtomicLongFieldUpdater<QueueImpl>
dispatchStartTimeUpdater = AtomicLongFieldUpdater.newUpdater(QueueImpl.class,
"dispatchStartTime");
@@ -2472,7 +2475,7 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
@Override
public synchronized boolean expireReference(final long messageID) throws
Exception {
- if (isExpirationRedundant()) {
+ if (isExpiryDisabled()) {
return false;
}
@@ -2494,7 +2497,7 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
@Override
public synchronized int expireReferences(final Filter filter) throws
Exception {
- if (isExpirationRedundant()) {
+ if (isExpiryDisabled()) {
return 0;
}
@@ -2523,14 +2526,13 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
@Override
public void expireReferences(Runnable done) {
- if (isExpirationRedundant()) {
+ if (isExpiryDisabled()) {
if (done != null) {
done.run();
}
return;
}
-
if (!queueDestroyed) {
getExecutor().execute(new ExpiryScanner(done));
} else {
@@ -2541,7 +2543,7 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
}
}
- public boolean isExpirationRedundant() {
+ private boolean isExpiryDisabled() {
final SimpleString expiryAddress = addressSettings.getExpiryAddress();
if (expiryAddress != null && expiryAddress.equals(this.address)) {
// check expire with itself would be silly (waste of time)
@@ -2550,6 +2552,11 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
return true;
}
+ if (isInternalQueue() && name.toString().startsWith(MIRROR_ADDRESS)) {
+ logger.trace("Mirror SNF queues are not supposed to expire messages.
Address={}, Queue={}", address, name);
+ return true;
+ }
+
return false;
}
diff --git
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPMirrorExpiryQueueTest.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPMirrorExpiryQueueTest.java
new file mode 100644
index 0000000000..5ff5ec7974
--- /dev/null
+++
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPMirrorExpiryQueueTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.connect;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.api.core.TransportConfiguration;
+import
org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration;
+import
org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
+import
org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPMirrorBrokerConnectionElement;
+import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.impl.QueueImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.RandomUtil;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class AMQPMirrorExpiryQueueTest extends ActiveMQTestBase {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final String EXPIRY_QUEUE =
AMQPMirrorExpiryQueueTest.class.getName() + "_ExpiryOut";
+
+ protected TransportConfiguration newAcceptorConfig(int port, String name) {
+ HashMap<String, Object> params = new HashMap<>();
+ params.put(TransportConstants.PORT_PROP_NAME, String.valueOf(port));
+ params.put(TransportConstants.PROTOCOLS_PROP_NAME, "AMQP,CORE,OPENWIRE");
+ HashMap<String, Object> amqpParams = new HashMap<>();
+ TransportConfiguration tc = new
TransportConfiguration(NETTY_ACCEPTOR_FACTORY, params, name, amqpParams);
+ return tc;
+ }
+
+ protected ActiveMQServer createServer(int port, String brokerName) throws
Exception {
+
+ final ActiveMQServer server = this.createServer(true, true);
+
+ server.getConfiguration().getAcceptorConfigurations().clear();
+
server.getConfiguration().getAcceptorConfigurations().add(newAcceptorConfig(port,
"netty-acceptor"));
+ server.getConfiguration().setName(brokerName);
+
server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory()
+ port);
+
server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory()
+ port);
+
server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory()
+ port);
+ server.getConfiguration().setJMXManagementEnabled(true);
+ server.getConfiguration().setMessageExpiryScanPeriod(5);
+
+ server.getConfiguration().addAddressSetting("#", new
AddressSettings().setExpiryAddress(SimpleString.of(EXPIRY_QUEUE)));
+ return server;
+ }
+
+ @Test
+ public void testExpiryOnMirrorSNF() throws Exception {
+
+ final long numberOfMessages = 100;
+ ActiveMQServer serverA = createServer(5671, getTestMethodName() + "_A");
+
+ {
+ AMQPBrokerConnectConfiguration amqpConnection = new
AMQPBrokerConnectConfiguration(getTestMethodName() + "_willNeverConnect",
"tcp://localhost:6671").setReconnectAttempts(1).setRetryInterval(10);
+ AMQPMirrorBrokerConnectionElement replica = new
AMQPMirrorBrokerConnectionElement().setType(AMQPBrokerConnectionAddressType.MIRROR).setDurable(true);
+ amqpConnection.addElement(replica);
+ serverA.getConfiguration().addAMQPConnection(amqpConnection);
+ }
+
+ String queueName = getTestMethodName() + "_" + RandomUtil.randomString();
+
+ serverA.setIdentity(getTestMethodName() + "_A");
+ serverA.start();
+
+
serverA.createQueue(QueueConfiguration.of(queueName).setName(queueName).setRoutingType(RoutingType.ANYCAST));
+ Queue expiryA =
serverA.createQueue(QueueConfiguration.of(EXPIRY_QUEUE).setName(EXPIRY_QUEUE).setRoutingType(RoutingType.ANYCAST));
+ Queue snfQueue = serverA.locateQueue(QueueImpl.MIRROR_ADDRESS + "_" +
getTestMethodName() + "_willNeverConnect");
+ assertNotNull(snfQueue);
+ assertNotNull(expiryA);
+
+ ConnectionFactory factoryA = CFUtil.createConnectionFactory("CORE",
"tcp://localhost:5671");
+ try (Connection connection = factoryA.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ MessageProducer producer =
session.createProducer(session.createQueue(queueName));
+ producer.setTimeToLive(1);
+
+ for (int i = 0; i < numberOfMessages; i++) {
+ producer.send(session.createTextMessage("hello" + i));
+ }
+
+ session.commit();
+ }
+
+ Thread.sleep(10); // waiting a little more than the expiry scan period
+ Wait.assertEquals(numberOfMessages, expiryA::getMessageCount, 5000, 100);
+
+ // We should still have the message sends and the acks from the expired
messages in the SNF
+ Wait.assertTrue(() -> snfQueue.getMessageCount() >= (numberOfMessages *
2), 5000, 100);
+ }
+
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
