You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2022/01/14 16:38:45 UTC
[activemq-artemis] branch 2.19.x updated (1f9b9be -> c11a0c0)
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a change to branch 2.19.x
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git.
from 1f9b9be NO-JIRA fixing intermittent failures
new a57cd75 ARTEMIS-3496 Replica connection to its live should fail fast
new 55658c2 ARTEMIS-3496 - add test to verify no reconnect on locators - mokito based and quite involved
new aeaf5bf NO-JIRA fixing intermittent test failure
new c11a0c0 ARTEMIS-3525 Empty Auto Created queues should be removed on restart
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../artemis/core/postoffice/AddressManager.java | 5 +
.../core/postoffice/impl/PostOfficeImpl.java | 20 +-
.../core/postoffice/impl/SimpleAddressManager.java | 9 +-
.../server/impl/ReplicationBackupActivation.java | 2 +-
.../server/impl/SharedNothingBackupActivation.java | 2 +-
.../postoffice/impl/PostOfficeTestAccessor.java | 2 +-
.../impl/BackupActivationNoReconnectTest.java | 278 +++++++++++++++++++++
.../tests/integration/client/AutoCreateTest.java | 91 +++++++
.../server/AutoCreateExpiryResourcesTest.java | 8 +-
9 files changed, 403 insertions(+), 14 deletions(-)
create mode 100644 artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/BackupActivationNoReconnectTest.java
[activemq-artemis] 03/04: NO-JIRA fixing intermittent test failure
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch 2.19.x
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit aeaf5bfdf5e36a32d4f435e904007f49d816ffde
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Oct 12 10:23:22 2021 -0400
NO-JIRA fixing intermittent test failure
(cherry picked from commit e86acc977b4f0c595156d423724d44624660085c)
---
.../tests/integration/server/AutoCreateExpiryResourcesTest.java | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateExpiryResourcesTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateExpiryResourcesTest.java
index ee88bc6..f6168dc 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateExpiryResourcesTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/AutoCreateExpiryResourcesTest.java
@@ -52,11 +52,10 @@ public class AutoCreateExpiryResourcesTest extends ActiveMQTestBase {
public void setUp() throws Exception {
super.setUp();
server = createServer(false);
- server.getConfiguration().setAddressQueueScanPeriod(100);
+ server.getConfiguration().setAddressQueueScanPeriod(50L).setMessageExpiryScanPeriod(50L);
// set common address settings needed for all tests; make sure to use getMatch instead of addMatch in invidual tests or these will be overwritten
server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateExpiryResources(true).setExpiryAddress(expiryAddress).setExpiryDelay(EXPIRY_DELAY));
- server.getConfiguration().setMessageExpiryScanPeriod(50L);
server.start();
}
@@ -164,9 +163,11 @@ public class AutoCreateExpiryResourcesTest extends ActiveMQTestBase {
ClientSessionFactory sessionFactory = createSessionFactory(locator);
ClientSession session = addClientSession(sessionFactory.createSession(true, true));
Wait.assertTrue(() -> server.locateQueue(expiryQueueName) != null, 2000, 100);
+ Wait.assertEquals(1, server.locateQueue(expiryQueueName) :: getMessageCount);
+
ClientConsumer consumer = session.createConsumer(expiryQueueName);
session.start();
- ClientMessage message = consumer.receive();
+ ClientMessage message = consumer.receive(5_000);
assertNotNull(message);
message.acknowledge();
consumer.close();
@@ -227,6 +228,7 @@ public class AutoCreateExpiryResourcesTest extends ActiveMQTestBase {
ClientSession session = addClientSession(sessionFactory.createSession(true, false));
ClientProducer producer = addClientProducer(session.createProducer(addressA));
ClientMessage message = session.createMessage(true);
+ message.setExpiration(System.currentTimeMillis());
producer.send(message);
producer.close();
session.close();
[activemq-artemis] 01/04: ARTEMIS-3496 Replica connection to its live should fail fast
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch 2.19.x
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit a57cd757f3e3ab6b07668ce964b55aa4fa5b0666
Author: franz1981 <ni...@gmail.com>
AuthorDate: Fri Sep 24 09:00:43 2021 +0200
ARTEMIS-3496 Replica connection to its live should fail fast
(cherry picked from commit 1dc9d4f45589a1e8d2565e23fb8cd0321241f6e9)
---
.../activemq/artemis/core/server/impl/ReplicationBackupActivation.java | 2 +-
.../artemis/core/server/impl/SharedNothingBackupActivation.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ReplicationBackupActivation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ReplicationBackupActivation.java
index 10212e0..35f5afe 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ReplicationBackupActivation.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ReplicationBackupActivation.java
@@ -566,7 +566,7 @@ public final class ReplicationBackupActivation extends Activation implements Dis
final ReplicationObserver liveObserver) {
ReplicationEndpoint replicationEndpoint = null;
try {
- liveControl.getSessionFactory().setReconnectAttempts(1);
+ liveControl.getSessionFactory().setReconnectAttempts(0);
liveObserver.listenConnectionFailuresOf(liveControl.getSessionFactory());
liveControl.authorize();
replicationEndpoint = new ReplicationEndpoint(activeMQServer, policy.isTryFailback(), liveObserver);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java
index 13421c8..85cbed6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java
@@ -498,7 +498,7 @@ public final class SharedNothingBackupActivation extends Activation implements R
public void run() {
try {
//we should only try once, if its not there we should move on.
- clusterControl.getSessionFactory().setReconnectAttempts(1);
+ clusterControl.getSessionFactory().setReconnectAttempts(0);
backupQuorum.setSessionFactory(clusterControl.getSessionFactory());
//get the connection and request replication to live
clusterControl.authorize();
[activemq-artemis] 02/04: ARTEMIS-3496 - add test to verify no reconnect on locators - mokito based and quite involved
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch 2.19.x
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 55658c23ca0a0bd312cc183d952265ce3e774058
Author: gtully <ga...@gmail.com>
AuthorDate: Fri Oct 8 16:53:06 2021 +0100
ARTEMIS-3496 - add test to verify no reconnect on locators - mokito based and quite involved
(cherry picked from commit 6f4c609ee0dc1d6671c7c9254991cd8bb895bea4)
---
.../impl/BackupActivationNoReconnectTest.java | 278 +++++++++++++++++++++
1 file changed, 278 insertions(+)
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/BackupActivationNoReconnectTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/BackupActivationNoReconnectTest.java
new file mode 100644
index 0000000..d19f38e
--- /dev/null
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/BackupActivationNoReconnectTest.java
@@ -0,0 +1,278 @@
+/*
+ *
+ * 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.core.server.impl;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.activemq.artemis.api.config.ServerLocatorConfig;
+import org.apache.activemq.artemis.api.core.ActiveMQDisconnectedException;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.TransportConfiguration;
+import org.apache.activemq.artemis.api.core.client.SessionFailureListener;
+import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryImpl;
+import org.apache.activemq.artemis.core.client.impl.ServerLocatorInternal;
+import org.apache.activemq.artemis.core.client.impl.TopologyMemberImpl;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.paging.PagingManager;
+import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.protocol.core.Channel;
+import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
+import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ClusterConnectReplyMessage;
+import org.apache.activemq.artemis.core.remoting.FailureListener;
+import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory;
+import org.apache.activemq.artemis.core.server.LiveNodeLocator;
+import org.apache.activemq.artemis.core.server.NodeManager;
+import org.apache.activemq.artemis.core.server.cluster.BackupManager;
+import org.apache.activemq.artemis.core.server.cluster.ClusterControl;
+import org.apache.activemq.artemis.core.server.cluster.ClusterController;
+import org.apache.activemq.artemis.core.server.cluster.ClusterManager;
+import org.apache.activemq.artemis.core.server.cluster.ha.HAPolicy;
+import org.apache.activemq.artemis.core.server.cluster.ha.ReplicaPolicy;
+import org.apache.activemq.artemis.core.server.cluster.ha.ReplicationBackupPolicy;
+import org.apache.activemq.artemis.core.server.cluster.ha.ReplicationPrimaryPolicy;
+import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager;
+import org.apache.activemq.artemis.core.server.cluster.qourum.SharedNothingBackupQuorum;
+import org.apache.activemq.artemis.core.server.management.ManagementService;
+import org.apache.activemq.artemis.core.version.Version;
+import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
+import org.apache.activemq.artemis.spi.core.remoting.ClientProtocolManager;
+import org.apache.activemq.artemis.utils.ExecutorFactory;
+import org.apache.activemq.artemis.utils.ThreadDumpUtil;
+import org.apache.activemq.artemis.utils.actors.ArtemisExecutor;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyByte;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.when;
+
+public class BackupActivationNoReconnectTest {
+
+ @Test(timeout = 30000)
+ public void verifyReplicationBackupActivation() throws Exception {
+ ReplicationBackupPolicy policy = Mockito.mock(ReplicationBackupPolicy.class);
+ ReplicationPrimaryPolicy replicationPrimaryPolicy = Mockito.mock(ReplicationPrimaryPolicy.class);
+ when(policy.getLivePolicy()).thenReturn(replicationPrimaryPolicy);
+ ActiveMQServerImpl server = Mockito.mock(ActiveMQServerImpl.class);
+
+ DistributedPrimitiveManager distributedManager = Mockito.mock(DistributedPrimitiveManager.class);
+ ReplicationBackupActivation replicationBackupActivation = new ReplicationBackupActivation(server, distributedManager, policy);
+
+ verifySingleAttemptToLocateLive(server, replicationBackupActivation);
+ }
+
+ @Test(timeout = 30000)
+ public void verifySharedNothingBackupActivation() throws Exception {
+ ActiveMQServerImpl server = Mockito.mock(ActiveMQServerImpl.class);
+ when(server.isStarted()).thenReturn(true);
+ ExecutorService threadPool = Mockito.mock(ExecutorService.class);
+ doAnswer(invocation -> {
+ Runnable runnable = invocation.getArgument(0);
+ runnable.run();
+ return null;
+ }).when(threadPool).execute(any(Runnable.class));
+ when(server.getThreadPool()).thenReturn(threadPool);
+ HAPolicy haPolicy = Mockito.mock(HAPolicy.class);
+ when(haPolicy.isBackup()).thenReturn(true);
+ when(server.getHAPolicy()).thenReturn(haPolicy);
+ when(server.getManagementService()).thenReturn(Mockito.mock(ManagementService.class));
+
+ Map<String, Object> activationParams = new HashMap<>();
+ ReplicaPolicy replicaPolicy = new ReplicaPolicy(null, 0);
+ replicaPolicy.setAllowFailback(false);
+
+ verifySingleAttemptToLocateLive(server, replicaPolicy.createActivation(server, false, activationParams, null));
+ }
+
+ protected void verifySingleAttemptToLocateLive(ActiveMQServerImpl server, Activation activation) throws Exception {
+
+ NodeManager nodeManager = Mockito.mock(NodeManager.class);
+ when(server.getNodeManager()).thenReturn(nodeManager);
+
+ when(nodeManager.getNodeActivationSequence()).thenReturn(0L);
+ when(server.initialisePart1(Mockito.anyBoolean())).thenReturn(true);
+
+ ClusterManager clusterManager = Mockito.mock(ClusterManager.class);
+ ClusterController clusterController = Mockito.mock(ClusterController.class);
+ when(clusterManager.getClusterController()).thenReturn(clusterController);
+ when(server.getClusterManager()).thenReturn(clusterManager);
+
+ QuorumManager quorumManager = Mockito.mock(QuorumManager.class);
+ when(clusterManager.getQuorumManager()).thenReturn(quorumManager);
+ Version version = Mockito.mock(Version.class);
+ when(server.getVersion()).thenReturn(version);
+ BackupManager backupManager = Mockito.mock(BackupManager.class);
+ when(server.getBackupManager()).thenReturn(backupManager);
+
+
+ final CountDownLatch gotLocator = new CountDownLatch(1);
+ final AtomicReference<LiveNodeLocator> locatorAtomicReference = new AtomicReference<>();
+ doAnswer(invocation -> {
+ locatorAtomicReference.set((LiveNodeLocator) invocation.getArguments()[0]);
+ gotLocator.countDown();
+ return null;
+ }).when(clusterController).addClusterTopologyListenerForReplication(Mockito.any());
+
+ when(server.checkLiveIsNotColocated(Mockito.anyString())).thenReturn(true);
+
+ ClusterControl clusterControl = Mockito.mock(ClusterControl.class);
+ when(clusterController.connectToNodeInReplicatedCluster(Mockito.any())).thenReturn(clusterControl);
+
+ // start listener that will accept replication channel
+ try (ServerSocket serverSocket = new ServerSocket(0)) {
+
+ final ServerLocatorInternal serverLocator = Mockito.mock(ServerLocatorInternal.class);
+ final TransportConfiguration connectorConfig = Mockito.mock(TransportConfiguration.class);
+ final ServerLocatorConfig locatorConfig = Mockito.mock(ServerLocatorConfig.class);
+ final int reconnectAttempts = 1;
+ final Executor threadPool = Mockito.mock(Executor.class);
+ final ScheduledExecutorService scheduledThreadPool = Mockito.mock(ScheduledExecutorService.class);
+ final ClientProtocolManager clientProtocolManager = Mockito.mock(ClientProtocolManager.class);
+ when(serverLocator.newProtocolManager()).thenReturn(clientProtocolManager);
+ when(connectorConfig.getFactoryClassName()).thenReturn(NettyConnectorFactory.class.getName());
+ Map<String, Object> urlParams = new HashMap<>();
+ urlParams.put("port", serverSocket.getLocalPort());
+ when(connectorConfig.getParams()).thenReturn(urlParams);
+ ClientSessionFactoryImpl sessionFactory = new ClientSessionFactoryImpl(serverLocator, connectorConfig, locatorConfig, reconnectAttempts, threadPool, scheduledThreadPool, null, null);
+ when(clusterControl.getSessionFactory()).thenReturn(sessionFactory);
+ when(clientProtocolManager.isAlive()).thenReturn(true);
+
+ CoreRemotingConnection remotingConnection = Mockito.mock(CoreRemotingConnection.class);
+ when(clientProtocolManager.connect(any(), anyLong(), anyLong(), any(), any(), any())).thenReturn(remotingConnection);
+ final CountDownLatch gotReplicationObserverListener = new CountDownLatch(1);
+ final CountDownLatch gotListener = new CountDownLatch(1);
+ AtomicReference<FailureListener> failureListenerAtomicReference = new AtomicReference<>();
+ doAnswer((Answer<Object>) invocation -> {
+ final FailureListener listener = invocation.getArgument(0);
+ if (listener instanceof ReplicationObserver || listener instanceof SharedNothingBackupQuorum) {
+ gotReplicationObserverListener.countDown();
+ } else {
+ failureListenerAtomicReference.set(listener);
+ gotListener.countDown();
+ }
+ return null;
+ }).when(remotingConnection).addFailureListener(any());
+ when(remotingConnection.getID()).thenReturn("First");
+
+ final Channel channel = Mockito.mock(Channel.class);
+ ClusterConnectReplyMessage clusterConnectReplyMessage = Mockito.mock(ClusterConnectReplyMessage.class);
+ when(channel.sendBlocking(any(), anyByte())).thenReturn(clusterConnectReplyMessage);
+ when(clusterConnectReplyMessage.isAuthorized()).thenReturn(true);
+ ExecutorFactory executorFactory = Mockito.mock(ExecutorFactory.class);
+ when(server.getExecutorFactory()).thenReturn(executorFactory);
+ ArtemisExecutor executor = Mockito.mock(ArtemisExecutor.class);
+ doAnswer(invocation -> {
+ Runnable runnable = invocation.getArgument(0);
+ runnable.run();
+ return null;
+ }).when(executor).execute(any(Runnable.class));
+
+ when(executorFactory.getExecutor()).thenReturn(executor);
+ when(clusterControl.createReplicationChannel()).thenReturn(channel);
+
+ StorageManager storageManager = Mockito.mock(StorageManager.class);
+ when(server.getStorageManager()).thenReturn(storageManager);
+ Journal journal = Mockito.mock(Journal.class);
+ when(storageManager.getBindingsJournal()).thenReturn(journal);
+ when(storageManager.getMessageJournal()).thenReturn(journal);
+ when(server.createPagingManager()).thenReturn(Mockito.mock(PagingManager.class));
+
+
+ AtomicBoolean reconnectWorkOnRetry = new AtomicBoolean();
+ final ActiveMQDisconnectedException forcedError = new ActiveMQDisconnectedException("DD");
+ when(clientProtocolManager.cleanupBeforeFailover(any())).then(invocation -> {
+ assertEquals(forcedError, invocation.getArgument(0));
+ reconnectWorkOnRetry.set(true);
+ return true;
+ });
+
+ final CountDownLatch failure = new CountDownLatch(2);
+ AtomicBoolean failoverEventReported = new AtomicBoolean(true);
+ sessionFactory.addFailureListener(new SessionFailureListener() {
+ @Override
+ public void beforeReconnect(ActiveMQException exception) {
+ // debatable whether this should be called when there will be no reconnect!
+ failure.countDown();
+ }
+
+ @Override
+ public void connectionFailed(ActiveMQException exception, boolean failedOver) {
+ }
+
+ @Override
+ public void connectionFailed(ActiveMQException exception, boolean failedOver, String scaleDownTargetNodeID) {
+ failure.countDown();
+ failoverEventReported.set(failedOver);
+ }
+ });
+
+ // start replicator, the actual test!
+ Thread activationThread = new Thread(activation);
+ activationThread.start();
+
+ // signal nodeUp once locator is in place
+ assertTrue(gotLocator.await(5, TimeUnit.SECONDS));
+ locatorAtomicReference.get().nodeUP(new TopologyMemberImpl("nodeId", "backupGroupName", "", new TransportConfiguration(), new TransportConfiguration()), false);
+
+ Socket clientSocket = serverSocket.accept();
+
+ // wait till failure listener is installed, then close
+ assertTrue("Replication observer in play", gotReplicationObserverListener.await(5, TimeUnit.SECONDS));
+ clientSocket.close();
+
+ // propagate the failure
+ assertTrue(gotListener.await(5, TimeUnit.SECONDS));
+ failureListenerAtomicReference.get().connectionFailed(forcedError, false);
+
+ // wait for activation completion
+ activationThread.join(10000);
+ if (activationThread.isAlive()) {
+ String dump = ThreadDumpUtil.threadDump("Activation thread is still alive!");
+ activationThread.interrupt();
+ Assert.fail(dump);
+ }
+
+ assertTrue(failure.await(5, TimeUnit.SECONDS));
+ assertFalse(failoverEventReported.get());
+
+ // ensure no reconnect
+ assertFalse(reconnectWorkOnRetry.get());
+ }
+ }
+}
\ No newline at end of file
[activemq-artemis] 04/04: ARTEMIS-3525 Empty Auto Created queues should be removed on restart
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch 2.19.x
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit c11a0c000475d2769d9c91df36c56b25a407fda2
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Oct 12 12:03:29 2021 -0400
ARTEMIS-3525 Empty Auto Created queues should be removed on restart
(cherry picked from commit 2383aa0125320713b9a753668b203a878c24b2e0)
---
.../artemis/core/postoffice/AddressManager.java | 5 ++
.../core/postoffice/impl/PostOfficeImpl.java | 20 +++--
.../core/postoffice/impl/SimpleAddressManager.java | 9 ++-
.../postoffice/impl/PostOfficeTestAccessor.java | 2 +-
.../tests/integration/client/AutoCreateTest.java | 91 ++++++++++++++++++++++
5 files changed, 118 insertions(+), 9 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
index f60b9a1..e7720da 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/AddressManager.java
@@ -85,6 +85,11 @@ public interface AddressManager {
boolean checkAutoRemoveAddress(SimpleString address,
AddressInfo addressInfo,
+ AddressSettings settings,
+ boolean ignoreDelay) throws Exception;
+
+ boolean checkAutoRemoveAddress(SimpleString address,
+ AddressInfo addressInfo,
AddressSettings settings) throws Exception;
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index feaa7b1..ba15fc7 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -1779,6 +1779,8 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
@Override
public synchronized void startAddressQueueScanner() {
+ reapAddresses(true); // we need to check for empty auto-created queues before the acceptors are on
+ // empty auto-created queues and addresses should be removed right away
if (addressQueueReaperPeriod > 0) {
if (addressQueueReaperRunnable != null)
addressQueueReaperRunnable.stop();
@@ -1845,7 +1847,7 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
@Override
public void run() {
- reapAddresses();
+ reapAddresses(false);
}
}
@@ -1855,12 +1857,16 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
/** To be used by the AddressQueueReaper.
* It is also exposed for tests through PostOfficeTestAccessor */
- void reapAddresses() {
+ void reapAddresses(boolean initialCheck) {
getLocalQueues().forEach(queue -> {
- if (!queue.isInternalQueue() && QueueManagerImpl.isAutoDelete(queue) && QueueManagerImpl.consumerCountCheck(queue) && QueueManagerImpl.delayCheck(queue) && QueueManagerImpl.messageCountCheck(queue) && queueWasUsed(queue)) {
- if (queue.isSwept()) {
+ if (!queue.isInternalQueue() && QueueManagerImpl.isAutoDelete(queue) && QueueManagerImpl.consumerCountCheck(queue) && (initialCheck || QueueManagerImpl.delayCheck(queue)) && QueueManagerImpl.messageCountCheck(queue) && (initialCheck || queueWasUsed(queue))) {
+ if (initialCheck || queue.isSwept()) {
if (logger.isDebugEnabled()) {
- logger.debug("Removing queue " + queue.getName() + " after it being swept twice on reaping process");
+ if (initialCheck) {
+ logger.debug("Removing queue " + queue.getName() + " during the reload check");
+ } else {
+ logger.debug("Removing queue " + queue.getName() + " after it being swept twice on reaping process");
+ }
}
QueueManagerImpl.performAutoDeleteQueue(server, queue);
} else {
@@ -1878,8 +1884,8 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
AddressSettings settings = addressSettingsRepository.getMatch(address.toString());
try {
- if (addressManager.checkAutoRemoveAddress(address, addressInfo, settings)) {
- if (addressInfo.isSwept()) {
+ if (addressManager.checkAutoRemoveAddress(address, addressInfo, settings, initialCheck)) {
+ if (initialCheck || addressInfo.isSwept()) {
server.autoRemoveAddressInfo(address, null);
} else {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
index 5a4a94a..2e92300 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/SimpleAddressManager.java
@@ -369,7 +369,14 @@ public class SimpleAddressManager implements AddressManager {
public boolean checkAutoRemoveAddress(SimpleString address,
AddressInfo addressInfo,
AddressSettings settings) throws Exception {
- return settings.isAutoDeleteAddresses() && addressInfo != null && addressInfo.isAutoCreated() && !bindingsFactory.isAddressBound(address) && addressInfo.getBindingRemovedTimestamp() != -1 && (System.currentTimeMillis() - addressInfo.getBindingRemovedTimestamp() >= settings.getAutoDeleteAddressesDelay());
+ return checkAutoRemoveAddress(address, addressInfo, settings, false);
+ }
+
+ @Override
+ public boolean checkAutoRemoveAddress(SimpleString address,
+ AddressInfo addressInfo,
+ AddressSettings settings, boolean ignoreDelay) throws Exception {
+ return settings.isAutoDeleteAddresses() && addressInfo != null && addressInfo.isAutoCreated() && !bindingsFactory.isAddressBound(address) && (ignoreDelay || addressInfo.getBindingRemovedTimestamp() != -1 && (System.currentTimeMillis() - addressInfo.getBindingRemovedTimestamp() >= settings.getAutoDeleteAddressesDelay()));
}
@Override
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeTestAccessor.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeTestAccessor.java
index 6236de9..9e3ecb0 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeTestAccessor.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeTestAccessor.java
@@ -20,7 +20,7 @@ package org.apache.activemq.artemis.core.postoffice.impl;
public class PostOfficeTestAccessor {
public static void reapAddresses(PostOfficeImpl postOffice) {
- postOffice.reapAddresses();
+ postOffice.reapAddresses(false);
}
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoCreateTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoCreateTest.java
index fad1357..9a7db67 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoCreateTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AutoCreateTest.java
@@ -22,6 +22,7 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
+import javax.jms.TextMessage;
import javax.jms.Topic;
import java.util.concurrent.CountDownLatch;
@@ -42,6 +43,7 @@ import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
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.jboss.logging.Logger;
import org.junit.After;
@@ -327,6 +329,95 @@ public class AutoCreateTest extends ActiveMQTestBase {
}
}
+ @Test
+ public void testCleanupAfterRebootOpenWire() throws Exception {
+ testCleanupAfterReboot("OPENWIRE", false);
+ }
+
+ @Test
+ public void testCleanupAfterRebootCore() throws Exception {
+ // there is no need to duplicate the test between usedelay and not.
+ // doing it in one of the protocols should be enough
+ testCleanupAfterReboot("CORE", true);
+ }
+
+ @Test
+ public void testCleanupAfterRebootAMQP() throws Exception {
+ testCleanupAfterReboot("AMQP", false);
+ }
+
+ public void testCleanupAfterReboot(String protocol, boolean useDelay) throws Exception {
+
+ if (useDelay) {
+ // setting up a delay, to make things a bit more challenging
+ server.getAddressSettingsRepository().addMatch(getName(), new AddressSettings().setAutoCreateAddresses(true).setAutoDeleteAddressesDelay(TimeUnit.DAYS.toMillis(1)).setAutoDeleteQueuesDelay(TimeUnit.DAYS.toMillis(1)));
+ }
+
+ AssertionLoggerHandler.startCapture();
+ server.getConfiguration().setAddressQueueScanPeriod(-1); // disabling scanner, we will perform it manually
+ server.start();
+ String QUEUE_NAME = getName();
+
+ ConnectionFactory cf = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
+ try (Connection connection = cf.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+ }
+
+ AddressInfo info = server.getPostOffice().getAddressInfo(SimpleString.toSimpleString(QUEUE_NAME));
+ Assert.assertNotNull(info);
+ Assert.assertTrue(info.isAutoCreated());
+
+ server.stop();
+ server.start();
+
+ Assert.assertTrue(AssertionLoggerHandler.findText("AMQ224113"));
+ Assert.assertTrue(AssertionLoggerHandler.findText("AMQ224112"));
+
+ AssertionLoggerHandler.clear();
+
+ String randomString = "random " + RandomUtil.randomString();
+
+ try (Connection connection = cf.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageProducer producer = session.createProducer(queue);
+ producer.send(session.createTextMessage(randomString));
+ }
+
+ info = server.getPostOffice().getAddressInfo(SimpleString.toSimpleString(QUEUE_NAME));
+ Assert.assertNotNull(info);
+ Assert.assertTrue(info.isAutoCreated());
+
+ server.stop();
+ server.start();
+
+ Assert.assertFalse(AssertionLoggerHandler.findText("AMQ224113")); // this time around the queue had messages, it has to exist
+ Assert.assertFalse(AssertionLoggerHandler.findText("AMQ224112"));
+
+ info = server.getPostOffice().getAddressInfo(SimpleString.toSimpleString(QUEUE_NAME));
+ Assert.assertNotNull(info);
+ Assert.assertTrue(info.isAutoCreated());
+
+ { // just a namespace
+ org.apache.activemq.artemis.core.server.Queue serverQueue = server.locateQueue(QUEUE_NAME);
+ Wait.assertEquals(1, serverQueue::getMessageCount);
+ }
+
+
+ try (Connection connection = cf.createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageConsumer consumer = session.createConsumer(queue);
+ TextMessage message = (TextMessage)consumer.receive(5000);
+ Assert.assertEquals(randomString, message.getText());
+ }
+
+ }
+
}