[
https://issues.apache.org/jira/browse/ARTEMIS-4527?focusedWorklogId=901549&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-901549
]
ASF GitHub Bot logged work on ARTEMIS-4527:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 25/Jan/24 01:13
Start Date: 25/Jan/24 01:13
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4705:
URL: https://github.com/apache/activemq-artemis/pull/4705#discussion_r1465730187
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/ClusterNotificationsTest.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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.cluster.distribution;
+
+import javax.jms.Connection;
+import javax.jms.MessageListener;
+import javax.jms.Queue;
+import javax.jms.Session;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import
org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
+import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
+import org.junit.Test;
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.listener.AbstractJmsListeningContainer;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.util.backoff.FixedBackOff;
+
+public class ClusterNotificationsTest extends ClusterTestBase {
+
+ protected boolean isNetty() {
+ return true;
+ }
+
+ @Test
+ public void testClusterNotificationsContinuity() throws Exception {
+ final MessageLoadBalancingType lbType =
MessageLoadBalancingType.OFF_WITH_REDISTRIBUTION;
+ final String namePrefix = "TEST.QUEUE.";
+ final int queueCount = 500;
+
+ AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler();
+ runAfter(loggerHandler::close);
+
+ setupServer(0, isFileStorage(), isNetty());
+ setupServer(1, isFileStorage(), isNetty());
+ setupServer(2, isFileStorage(), isNetty());
+
+ setupClusterConnection("cluster0", "", lbType, 1, isNetty(), 0, 1, 2);
+ setupClusterConnection("cluster0", "", lbType, 1, isNetty(), 1, 2, 0);
+ setupClusterConnection("cluster0", "", lbType, 1, isNetty(), 2, 0, 1);
+
+ startServers(0, 1, 2);
+
+ AddressSettings as = new AddressSettings().setRedistributionDelay(0);
+ servers[0].getAddressSettingsRepository().addMatch(namePrefix + "#", as);
+ servers[1].getAddressSettingsRepository().addMatch(namePrefix + "#", as);
+ servers[2].getAddressSettingsRepository().addMatch(namePrefix + "#", as);
+
+ waitForTopology(servers[2], 3);
+
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ CountDownLatch latch = new CountDownLatch(queueCount * 2);
+
+ try (ActiveMQConnectionFactory factory = new
ActiveMQConnectionFactory("tcp://localhost:61616")) {
+
+ List<DefaultMessageListenerContainer> containers = new ArrayList<>();
+ runAfter(() ->
containers.parallelStream().forEach(AbstractJmsListeningContainer::shutdown));
+ runAfter(() -> executorService.shutdown());
+
+ for (int i = 0; i < queueCount; i++) {
+ Queue queue = ActiveMQDestination.createQueue(namePrefix + i +
"-");
+
+ executorService.submit(() -> {
+ try (Connection connection = factory.createConnection();
+ Session session =
connection.createSession(Session.SESSION_TRANSACTED)) {
+
+
session.createProducer(queue).send(session.createTextMessage("Message"));
+ session.commit();
+ latch.countDown();
+ } catch (Exception e) { }
+ });
+
+ DefaultMessageListenerContainer container = new
DefaultMessageListenerContainer();
+ container.setCacheLevelName("CACHE_NONE");
+ container.setSessionTransacted(true);
+ container.setSessionAcknowledgeModeName("SESSION_TRANSACTED");
+ container.setConcurrentConsumers(20);
+ container.setConnectionFactory(new
CachingConnectionFactory(factory));
+ container.setDestinationName(queue.getQueueName());
+ container.setBackOff(new FixedBackOff(1000, 2));
+ container.setReceiveTimeout(100);
+ container.setMessageListener((MessageListener) msg -> {
+ latch.countDown();
+ });
+
+ container.initialize();
Review Comment:
Ok. Let's make it a soak test. And merge this PR.
@AntonRoskvist let me know if you have any issues doing it. If you can't
let me know.
Issue Time Tracking
-------------------
Worklog Id: (was: 901549)
Time Spent: 3h 50m (was: 3h 40m)
> Redistributor race when consumerCount reaches 0 in cluster
> ----------------------------------------------------------
>
> Key: ARTEMIS-4527
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4527
> Project: ActiveMQ Artemis
> Issue Type: Bug
> Reporter: Anton Roskvist
> Priority: Major
> Time Spent: 3h 50m
> Remaining Estimate: 0h
>
> This is a very rare bug caused by cluster notifications arriving in the wrong
> order in some very specific circumstances
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
