[
https://issues.apache.org/jira/browse/ARTEMIS-4527?focusedWorklogId=901514&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-901514
]
ASF GitHub Bot logged work on ARTEMIS-4527:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 24/Jan/24 17:52
Start Date: 24/Jan/24 17:52
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4705:
URL: https://github.com/apache/activemq-artemis/pull/4705#discussion_r1465340237
##########
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:
This test is always failing for me:
```
java.lang.OutOfMemoryError: unable to create native thread: possibly out of
memory or process/resource limits reached
at java.base/java.lang.Thread.start0(Native Method)
at java.base/java.lang.Thread.start(Thread.java:802)
at
org.springframework.core.task.SimpleAsyncTaskExecutor.doExecute(SimpleAsyncTaskExecutor.java:248)
at
org.springframework.core.task.SimpleAsyncTaskExecutor.execute(SimpleAsyncTaskExecutor.java:202)
at
org.springframework.core.task.SimpleAsyncTaskExecutor.execute(SimpleAsyncTaskExecutor.java:180)
at
org.springframework.jms.listener.DefaultMessageListenerContainer.doRescheduleTask(DefaultMessageListenerContainer.java:780)
at
org.springframework.jms.listener.AbstractJmsListeningContainer.resumePausedTasks(AbstractJmsListeningContainer.java:545)
at
org.springframework.jms.listener.AbstractJmsListeningContainer.doStart(AbstractJmsListeningContainer.java:299)
at
org.springframework.jms.listener.AbstractJmsListeningContainer.start(AbstractJmsListeningContainer.java:277)
at
org.springframework.jms.listener.DefaultMessageListenerContainer.start(DefaultMessageListenerContainer.java:657)
at
org.apache.activemq.artemis.tests.integration.cluster.distribution.ClusterNotificationsTest.testClusterNotificationsContinuity(ClusterNotificationsTest.java:110)
at
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
at
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
at
org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:54)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:54)
at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:54)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
at
org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
at
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
at
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
at
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
at org.junit.rules.RunRules.evaluate(RunRules.java:20)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at
com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
at
com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38)
at
com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:11)
at
com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35)
at
com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:232)
at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:55)
```
Can we do this test without spring though? it gets easier to debug tests.
I'm not aware of any other tests actually using spring for consuming messages
and asserting.
I reckon it removes a lot of boilerplate code from the consumer.. but I
would prefer having more control on what's going on...
like now: the tests Is failing with lack of threads and I have no idea how
to fix it.
Issue Time Tracking
-------------------
Worklog Id: (was: 901514)
Time Spent: 2h 20m (was: 2h 10m)
> 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: 2h 20m
> 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)
