gemmellr commented on code in PR #4734:
URL: https://github.com/apache/activemq-artemis/pull/4734#discussion_r1445899444
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/MirrorControllerBasicTest.java:
##########
@@ -94,7 +94,7 @@ public void testDirectSend() throws Exception {
server.createQueue(new
QueueConfiguration("test").setAddress("test").setRoutingType(RoutingType.ANYCAST));
Message message = AMQPMirrorMessageFactory.createMessage("test",
SimpleString.toSimpleString("ad1"), SimpleString.toSimpleString("qu1"), "test",
"someUID", "body-test", AckReason.KILLED);
- AMQPMirrorControllerSource.route(server, message);
+ AMQPMirrorControllerSource.routeMirrorCommand(server, message);
Review Comment:
This change doesnt appear to be in the same commit as the later change that
actually renamed the method, so it seems like one or more of these commits
might not compile.
##########
tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/IdemPotentACKTest.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.soak.brokerConnection.mirror;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.TransactionRolledBackException;
+import java.io.File;
+import java.io.StringWriter;
+import java.lang.invoke.MethodHandles;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.activemq.artemis.api.core.management.SimpleManagement;
+import
org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
+import org.apache.activemq.artemis.tests.soak.SoakTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.util.ServerUtil;
+import org.apache.activemq.artemis.utils.cli.helper.HelperCreate;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IdemPotentACKTest extends SoakTestBase {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static String largeBody;
+
+ static {
+ StringWriter writer = new StringWriter();
+ while (writer.getBuffer().length() < 1024 * 1024) {
+ writer.append("This is a large string ..... ");
+ }
+ largeBody = writer.toString();
+ }
+
+ private static final String QUEUE_NAME = "myQueue";
+
+ public static final String DC1_NODE_A = "idemPotentMirror/DC1";
+ public static final String DC2_NODE_A = "idemPotentMirror/DC2";
Review Comment:
ditto
##########
tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/IdemPotentACKTest.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.soak.brokerConnection.mirror;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.TransactionRolledBackException;
+import java.io.File;
+import java.io.StringWriter;
+import java.lang.invoke.MethodHandles;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.activemq.artemis.api.core.management.SimpleManagement;
+import
org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
+import org.apache.activemq.artemis.tests.soak.SoakTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.util.ServerUtil;
+import org.apache.activemq.artemis.utils.cli.helper.HelperCreate;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IdemPotentACKTest extends SoakTestBase {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static String largeBody;
+
+ static {
+ StringWriter writer = new StringWriter();
+ while (writer.getBuffer().length() < 1024 * 1024) {
+ writer.append("This is a large string ..... ");
+ }
+ largeBody = writer.toString();
+ }
+
+ private static final String QUEUE_NAME = "myQueue";
+
+ public static final String DC1_NODE_A = "idemPotentMirror/DC1";
+ public static final String DC2_NODE_A = "idemPotentMirror/DC2";
+
+ Process processDC1_node_A;
+ Process processDC2_node_A;
+
+ private static String DC1_NODEA_URI = "tcp://localhost:61616";
+ private static String DC2_NODEA_URI = "tcp://localhost:61618";
+
+ private static void createServer(String serverName,
+ String connectionName,
+ String mirrorURI,
+ int porOffset) throws Exception {
+ File serverLocation = getFileServerLocation(serverName);
+ deleteDirectory(serverLocation);
+
+ HelperCreate cliCreateServer = new HelperCreate();
+
cliCreateServer.setAllowAnonymous(true).setNoWeb(true).setArtemisInstance(serverLocation);
+ cliCreateServer.setMessageLoadBalancing("ON_DEMAND");
+ cliCreateServer.setClustered(false);
+ cliCreateServer.setNoWeb(true);
+ cliCreateServer.setArgs("--no-stomp-acceptor", "--no-hornetq-acceptor",
"--no-mqtt-acceptor", "--no-amqp-acceptor", "--max-hops", "1", "--name",
DC1_NODE_A);
+ cliCreateServer.addArgs("--queues", QUEUE_NAME);
+ cliCreateServer.setPortOffset(porOffset);
+ cliCreateServer.createServer();
+
+ Properties brokerProperties = new Properties();
+ brokerProperties.put("AMQPConnections." + connectionName + ".uri",
mirrorURI);
+ brokerProperties.put("AMQPConnections." + connectionName +
".retryInterval", "1000");
+ brokerProperties.put("AMQPConnections." + connectionName + ".type",
AMQPBrokerConnectionAddressType.MIRROR.toString());
+ brokerProperties.put("AMQPConnections." + connectionName +
".connectionElements.mirror.sync", "false");
+ brokerProperties.put("largeMessageSync", "false");
+ File brokerPropertiesFile = new File(serverLocation,
"broker.properties");
+ saveProperties(brokerProperties, brokerPropertiesFile);
+ }
+
+ @BeforeClass
+ public static void createServers() throws Exception {
+ createServer(DC1_NODE_A, "mirror", DC2_NODEA_URI, 0);
+ createServer(DC2_NODE_A, "mirror", DC1_NODEA_URI, 2);
+ }
+
+ private void startServers() throws Exception {
+ processDC1_node_A = startServer(DC1_NODE_A, -1, -1, new
File(getServerLocation(DC1_NODE_A), "broker.properties"));
+ processDC2_node_A = startServer(DC2_NODE_A, -1, -1, new
File(getServerLocation(DC2_NODE_A), "broker.properties"));
+
+ ServerUtil.waitForServerToStart(0, 10_000);
+ ServerUtil.waitForServerToStart(2, 10_000);
+ }
+
+ @Before
+ public void cleanupServers() {
+ cleanupData(DC1_NODE_A);
+ cleanupData(DC2_NODE_A);
+ }
+
+ private void transactSend(Session session, MessageProducer producer, int
initialCounter, int numberOfMessages) throws Throwable {
+ try {
+ for (int i = initialCounter; i < initialCounter + numberOfMessages;
i++) {
+ TextMessage message;
+ message = session.createTextMessage(largeBody);
+ message.setIntProperty("i", i);
+ message.setBooleanProperty("large", true);
+ String unique = "Unique " + i;
+
message.setStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_DUPLICATE_DETECTION_ID.toString(),
unique);
+ producer.send(message);
+ }
+ session.commit();
+ } catch (JMSException e) {
+ if (e instanceof TransactionRolledBackException &&
e.getMessage().contains("Duplicate message detected")) {
+ logger.debug("OK Exception {}", e.getMessage(), e);
+ return; // ok
+ } else {
+ logger.warn("Not OK Exception {}", e.getMessage(), e);
+ throw e;
+ }
+ }
+ }
+
+
+ @Test
+ public void testAMQP() throws Exception {
+ testACKs("AMQP");
+ }
+
+ @Test
+ public void testCORE() throws Exception {
+ testACKs("CORE");
+ }
+
+ private void testACKs(final String protocol) throws Exception {
+ startServers();
+
+ final int consumers = 10;
+ final int numberOfMessages = 300;
+ final int messagesPerConsumer = numberOfMessages / consumers;
+
+ // Just a reminder: if you change number on this test, this needs to be
true:
+ Assert.assertEquals("Invalid test config", 0, numberOfMessages %
consumers);
+
+ AtomicBoolean running = new AtomicBoolean(true);
+ runAfter(() -> running.set(false));
+
+ String snfQueue = "$ACTIVEMQ_ARTEMIS_MIRROR_mirror";
+
+ ExecutorService executor = Executors.newFixedThreadPool(consumers);
+ runAfter(executor::shutdownNow);
+ ConnectionFactory connectionFactoryDC1A =
CFUtil.createConnectionFactory(protocol, DC1_NODEA_URI);
+ CountDownLatch sendDone = new CountDownLatch(1);
+ CountDownLatch killSend = new CountDownLatch(1);
+
+ executor.execute(() -> {
+ int messagesSent = 0;
+ while (running.get() && messagesSent < numberOfMessages) {
+ try (Connection connection =
connectionFactoryDC1A.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageProducer producer = session.createProducer(queue);
+ if (messagesSent < 100) {
+ transactSend(session, producer, messagesSent, 1);
+ messagesSent++;
+ logger.debug("Sent {}", messagesSent);
+ if (messagesSent == 100) {
+ logger.debug("Signal to kill");
+ killSend.countDown();
+ }
+ } else {
+ transactSend(session, producer, messagesSent, 100);
+ messagesSent += 100;
+ logger.debug("Sent {}", messagesSent);
+ }
+ } catch (Throwable e) {
+ logger.debug(e.getMessage(), e);
+ try {
+ Thread.sleep(100);
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ sendDone.countDown();
+ });
+
+ Assert.assertTrue(killSend.await(10, TimeUnit.SECONDS));
+
+ restartDC1_ServerA();
+
+ Assert.assertTrue(sendDone.await(50, TimeUnit.SECONDS));
+
+ SimpleManagement simpleManagementDC1A = new
SimpleManagement(DC1_NODEA_URI, null, null);
+ SimpleManagement simpleManagementDC2A = new
SimpleManagement(DC2_NODEA_URI, null, null);
+
+ Wait.assertEquals(0, () -> getCount(simpleManagementDC1A, snfQueue));
+ Wait.assertEquals(numberOfMessages, () -> getCount(simpleManagementDC1A,
QUEUE_NAME));
+ Wait.assertEquals(numberOfMessages, () -> getCount(simpleManagementDC2A,
QUEUE_NAME));
+
+ AtomicInteger errors = new AtomicInteger(0);
+
+ CountDownLatch latchKill = new CountDownLatch(consumers);
+
+ CountDownLatch latchDone = new CountDownLatch(consumers);
+
+ Runnable runnableConsumer = () -> {
+ int messagesConsumed = 0;
+ while (running.get() && messagesConsumed < messagesPerConsumer) {
+ try (Connection connection =
connectionFactoryDC1A.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+ while (messagesConsumed < messagesPerConsumer) {
+ Message message = consumer.receive(100);
+ if (message == null || !(message instanceof TextMessage)) {
Review Comment:
the null check is redundant since given the instanceof check
##########
tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/IdemPotentACKTest.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.soak.brokerConnection.mirror;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.TransactionRolledBackException;
+import java.io.File;
+import java.io.StringWriter;
+import java.lang.invoke.MethodHandles;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.activemq.artemis.api.core.management.SimpleManagement;
+import
org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
+import org.apache.activemq.artemis.tests.soak.SoakTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.util.ServerUtil;
+import org.apache.activemq.artemis.utils.cli.helper.HelperCreate;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IdemPotentACKTest extends SoakTestBase {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static String largeBody;
+
+ static {
+ StringWriter writer = new StringWriter();
+ while (writer.getBuffer().length() < 1024 * 1024) {
+ writer.append("This is a large string ..... ");
+ }
+ largeBody = writer.toString();
+ }
+
+ private static final String QUEUE_NAME = "myQueue";
+
+ public static final String DC1_NODE_A = "idemPotentMirror/DC1";
+ public static final String DC2_NODE_A = "idemPotentMirror/DC2";
+
+ Process processDC1_node_A;
+ Process processDC2_node_A;
+
+ private static String DC1_NODEA_URI = "tcp://localhost:61616";
+ private static String DC2_NODEA_URI = "tcp://localhost:61618";
+
+ private static void createServer(String serverName,
+ String connectionName,
+ String mirrorURI,
+ int porOffset) throws Exception {
+ File serverLocation = getFileServerLocation(serverName);
+ deleteDirectory(serverLocation);
+
+ HelperCreate cliCreateServer = new HelperCreate();
+
cliCreateServer.setAllowAnonymous(true).setNoWeb(true).setArtemisInstance(serverLocation);
+ cliCreateServer.setMessageLoadBalancing("ON_DEMAND");
+ cliCreateServer.setClustered(false);
+ cliCreateServer.setNoWeb(true);
+ cliCreateServer.setArgs("--no-stomp-acceptor", "--no-hornetq-acceptor",
"--no-mqtt-acceptor", "--no-amqp-acceptor", "--max-hops", "1", "--name",
DC1_NODE_A);
+ cliCreateServer.addArgs("--queues", QUEUE_NAME);
+ cliCreateServer.setPortOffset(porOffset);
+ cliCreateServer.createServer();
+
+ Properties brokerProperties = new Properties();
+ brokerProperties.put("AMQPConnections." + connectionName + ".uri",
mirrorURI);
+ brokerProperties.put("AMQPConnections." + connectionName +
".retryInterval", "1000");
+ brokerProperties.put("AMQPConnections." + connectionName + ".type",
AMQPBrokerConnectionAddressType.MIRROR.toString());
+ brokerProperties.put("AMQPConnections." + connectionName +
".connectionElements.mirror.sync", "false");
+ brokerProperties.put("largeMessageSync", "false");
+ File brokerPropertiesFile = new File(serverLocation,
"broker.properties");
+ saveProperties(brokerProperties, brokerPropertiesFile);
+ }
+
+ @BeforeClass
+ public static void createServers() throws Exception {
+ createServer(DC1_NODE_A, "mirror", DC2_NODEA_URI, 0);
+ createServer(DC2_NODE_A, "mirror", DC1_NODEA_URI, 2);
+ }
+
+ private void startServers() throws Exception {
+ processDC1_node_A = startServer(DC1_NODE_A, -1, -1, new
File(getServerLocation(DC1_NODE_A), "broker.properties"));
+ processDC2_node_A = startServer(DC2_NODE_A, -1, -1, new
File(getServerLocation(DC2_NODE_A), "broker.properties"));
+
+ ServerUtil.waitForServerToStart(0, 10_000);
+ ServerUtil.waitForServerToStart(2, 10_000);
+ }
+
+ @Before
+ public void cleanupServers() {
+ cleanupData(DC1_NODE_A);
+ cleanupData(DC2_NODE_A);
+ }
+
+ private void transactSend(Session session, MessageProducer producer, int
initialCounter, int numberOfMessages) throws Throwable {
+ try {
+ for (int i = initialCounter; i < initialCounter + numberOfMessages;
i++) {
+ TextMessage message;
+ message = session.createTextMessage(largeBody);
+ message.setIntProperty("i", i);
+ message.setBooleanProperty("large", true);
+ String unique = "Unique " + i;
+
message.setStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_DUPLICATE_DETECTION_ID.toString(),
unique);
+ producer.send(message);
+ }
+ session.commit();
+ } catch (JMSException e) {
+ if (e instanceof TransactionRolledBackException &&
e.getMessage().contains("Duplicate message detected")) {
+ logger.debug("OK Exception {}", e.getMessage(), e);
+ return; // ok
+ } else {
+ logger.warn("Not OK Exception {}", e.getMessage(), e);
+ throw e;
+ }
+ }
+ }
+
+
+ @Test
+ public void testAMQP() throws Exception {
+ testACKs("AMQP");
+ }
+
+ @Test
+ public void testCORE() throws Exception {
+ testACKs("CORE");
+ }
+
+ private void testACKs(final String protocol) throws Exception {
+ startServers();
+
+ final int consumers = 10;
+ final int numberOfMessages = 300;
+ final int messagesPerConsumer = numberOfMessages / consumers;
+
+ // Just a reminder: if you change number on this test, this needs to be
true:
+ Assert.assertEquals("Invalid test config", 0, numberOfMessages %
consumers);
+
+ AtomicBoolean running = new AtomicBoolean(true);
+ runAfter(() -> running.set(false));
+
+ String snfQueue = "$ACTIVEMQ_ARTEMIS_MIRROR_mirror";
+
+ ExecutorService executor = Executors.newFixedThreadPool(consumers);
+ runAfter(executor::shutdownNow);
+ ConnectionFactory connectionFactoryDC1A =
CFUtil.createConnectionFactory(protocol, DC1_NODEA_URI);
+ CountDownLatch sendDone = new CountDownLatch(1);
+ CountDownLatch killSend = new CountDownLatch(1);
+
+ executor.execute(() -> {
+ int messagesSent = 0;
+ while (running.get() && messagesSent < numberOfMessages) {
+ try (Connection connection =
connectionFactoryDC1A.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageProducer producer = session.createProducer(queue);
+ if (messagesSent < 100) {
+ transactSend(session, producer, messagesSent, 1);
+ messagesSent++;
+ logger.debug("Sent {}", messagesSent);
+ if (messagesSent == 100) {
+ logger.debug("Signal to kill");
+ killSend.countDown();
+ }
+ } else {
+ transactSend(session, producer, messagesSent, 100);
+ messagesSent += 100;
+ logger.debug("Sent {}", messagesSent);
+ }
+ } catch (Throwable e) {
+ logger.debug(e.getMessage(), e);
+ try {
+ Thread.sleep(100);
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ sendDone.countDown();
+ });
+
+ Assert.assertTrue(killSend.await(10, TimeUnit.SECONDS));
+
+ restartDC1_ServerA();
+
+ Assert.assertTrue(sendDone.await(50, TimeUnit.SECONDS));
+
+ SimpleManagement simpleManagementDC1A = new
SimpleManagement(DC1_NODEA_URI, null, null);
+ SimpleManagement simpleManagementDC2A = new
SimpleManagement(DC2_NODEA_URI, null, null);
+
+ Wait.assertEquals(0, () -> getCount(simpleManagementDC1A, snfQueue));
+ Wait.assertEquals(numberOfMessages, () -> getCount(simpleManagementDC1A,
QUEUE_NAME));
+ Wait.assertEquals(numberOfMessages, () -> getCount(simpleManagementDC2A,
QUEUE_NAME));
+
+ AtomicInteger errors = new AtomicInteger(0);
+
+ CountDownLatch latchKill = new CountDownLatch(consumers);
+
+ CountDownLatch latchDone = new CountDownLatch(consumers);
+
+ Runnable runnableConsumer = () -> {
+ int messagesConsumed = 0;
+ while (running.get() && messagesConsumed < messagesPerConsumer) {
+ try (Connection connection =
connectionFactoryDC1A.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+ while (messagesConsumed < messagesPerConsumer) {
+ Message message = consumer.receive(100);
+ if (message == null || !(message instanceof TextMessage)) {
+ logger.debug("message received={}", message);
+ errors.incrementAndGet();
Review Comment:
This is tracking errors, but I dont see it being checked anywhere, and it
aso still proceeds to increment the messageConsumed count regardless...so it
doesnt care what it consumes, or if it consumed anything at all?
##########
tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/IdemPotentACKTest.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.soak.brokerConnection.mirror;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.TransactionRolledBackException;
+import java.io.File;
+import java.io.StringWriter;
+import java.lang.invoke.MethodHandles;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.activemq.artemis.api.core.management.SimpleManagement;
+import
org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
+import org.apache.activemq.artemis.tests.soak.SoakTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.util.ServerUtil;
+import org.apache.activemq.artemis.utils.cli.helper.HelperCreate;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IdemPotentACKTest extends SoakTestBase {
Review Comment:
IdemPotent - > Idempotent, one word.
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -213,15 +215,31 @@ public void deleteQueue(SimpleString address,
SimpleString queue) throws Excepti
if (deleteQueues) {
Message message = createMessage(address, queue, DELETE_QUEUE, null,
queue.toString());
- route(server, message);
+ routeMirrorCommand(server, message);
}
}
+ private boolean invalidTarget(MirrorController controller, Message message)
{
+ if (controller == null) {
+ return false;
+ }
+ String remoteID = getRemoteMirrorId();
+ if (remoteID == null) {
+ // This is to avoid a reflection from a small period of time one node
reconnects but not the opposite direction
+ remoteID =
String.valueOf(message.getAnnotation(BROKER_ID_SIMPLE_STRING));
+ logger.debug("Remote link is not initialized yet, setting remoteID
from message as {}", remoteID);
+ }
+ return sameNode(remoteID, controller.getRemoteMirrorId());
+ }
+
private boolean invalidTarget(MirrorController controller) {
return controller != null && sameNode(getRemoteMirrorId(),
controller.getRemoteMirrorId());
}
private boolean ignoreAddress(SimpleString address) {
+ if
(address.startsWith(server.getConfiguration().getManagementAddress())) {
+ return true;
+ }
Review Comment:
Seems like perhaps a separate change?
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -213,15 +215,31 @@ public void deleteQueue(SimpleString address,
SimpleString queue) throws Excepti
if (deleteQueues) {
Message message = createMessage(address, queue, DELETE_QUEUE, null,
queue.toString());
- route(server, message);
+ routeMirrorCommand(server, message);
}
}
+ private boolean invalidTarget(MirrorController controller, Message message)
{
+ if (controller == null) {
+ return false;
+ }
+ String remoteID = getRemoteMirrorId();
+ if (remoteID == null) {
+ // This is to avoid a reflection from a small period of time one node
reconnects but not the opposite direction
+ remoteID =
String.valueOf(message.getAnnotation(BROKER_ID_SIMPLE_STRING));
+ logger.debug("Remote link is not initialized yet, setting remoteID
from message as {}", remoteID);
Review Comment:
Worth null check before valueOf and assigning and logging, given many
messages coming through here probably wont have the annotation?
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -609,10 +600,22 @@ private Message createMessage(SimpleString address,
SimpleString queue, Object e
return
AMQPMirrorMessageFactory.createMessage(snfQueue.getAddress().toString(),
address, queue, event, brokerID, body, ackReason);
}
- public static void route(ActiveMQServer server, Message message) throws
Exception {
+ public static void routeMirrorCommand(ActiveMQServer server, Message
message) throws Exception {
+ message.setMessageID(server.getStorageManager().generateID());
+ RoutingContext ctx = mirrorControlRouting.get();
+ // it is important to use local only at the source to avoid having the
message strictly load balancing
+ // to other nodes if the SNF queue has the same name as the one on this
node.
+
ctx.clear().setMirrorOption(MirrorOption.disabled).setLoadBalancingType(MessageLoadBalancingType.LOCAL_ONLY);
+ server.getPostOffice().route(message, ctx, false);
+ }
Review Comment:
Could the impl just be a shared with the other near-duplicate method impl
below, and simply null-check the tx param before setting it (if you even need
to do that) ?
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/AMQPBrokerConnection.java:
##########
@@ -498,7 +498,7 @@ private Queue
installMirrorController(AMQPMirrorBrokerConnectionElement replicaC
Queue mirrorControlQueue =
server.locateQueue(getMirrorSNF(replicaConfig));
if (mirrorControlQueue == null) {
- mirrorControlQueue = server.createQueue(new
QueueConfiguration(getMirrorSNF(replicaConfig)).setAddress(getMirrorSNF(replicaConfig)).setRoutingType(RoutingType.ANYCAST).setDurable(replicaConfig.isDurable()).setInternal(true),
true);
+ mirrorControlQueue = server.createQueue(new
QueueConfiguration(getMirrorSNF(replicaConfig)).setAddress(getMirrorSNF(replicaConfig)).setRoutingType(RoutingType.ANYCAST).setDurable(replicaConfig.isDurable()).setInternal(false),
true);
Review Comment:
Though wasnt it explicitly made 'internal' when adding the 'multi mirror'
stuff originally? I think it was in part to aid ignoring it for some stuff
(besides management). Are those desired behaviours still being handled
otherwise now?
##########
tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/IdemPotentACKTest.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.soak.brokerConnection.mirror;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.TransactionRolledBackException;
+import java.io.File;
+import java.io.StringWriter;
+import java.lang.invoke.MethodHandles;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.activemq.artemis.api.core.management.SimpleManagement;
+import
org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
+import org.apache.activemq.artemis.tests.soak.SoakTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.apache.activemq.artemis.util.ServerUtil;
+import org.apache.activemq.artemis.utils.cli.helper.HelperCreate;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class IdemPotentACKTest extends SoakTestBase {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static String largeBody;
+
+ static {
+ StringWriter writer = new StringWriter();
+ while (writer.getBuffer().length() < 1024 * 1024) {
+ writer.append("This is a large string ..... ");
+ }
+ largeBody = writer.toString();
+ }
+
+ private static final String QUEUE_NAME = "myQueue";
+
+ public static final String DC1_NODE_A = "idemPotentMirror/DC1";
+ public static final String DC2_NODE_A = "idemPotentMirror/DC2";
+
+ Process processDC1_node_A;
+ Process processDC2_node_A;
+
+ private static String DC1_NODEA_URI = "tcp://localhost:61616";
+ private static String DC2_NODEA_URI = "tcp://localhost:61618";
+
+ private static void createServer(String serverName,
+ String connectionName,
+ String mirrorURI,
+ int porOffset) throws Exception {
+ File serverLocation = getFileServerLocation(serverName);
+ deleteDirectory(serverLocation);
+
+ HelperCreate cliCreateServer = new HelperCreate();
+
cliCreateServer.setAllowAnonymous(true).setNoWeb(true).setArtemisInstance(serverLocation);
+ cliCreateServer.setMessageLoadBalancing("ON_DEMAND");
+ cliCreateServer.setClustered(false);
+ cliCreateServer.setNoWeb(true);
+ cliCreateServer.setArgs("--no-stomp-acceptor", "--no-hornetq-acceptor",
"--no-mqtt-acceptor", "--no-amqp-acceptor", "--max-hops", "1", "--name",
DC1_NODE_A);
+ cliCreateServer.addArgs("--queues", QUEUE_NAME);
+ cliCreateServer.setPortOffset(porOffset);
+ cliCreateServer.createServer();
+
+ Properties brokerProperties = new Properties();
+ brokerProperties.put("AMQPConnections." + connectionName + ".uri",
mirrorURI);
+ brokerProperties.put("AMQPConnections." + connectionName +
".retryInterval", "1000");
+ brokerProperties.put("AMQPConnections." + connectionName + ".type",
AMQPBrokerConnectionAddressType.MIRROR.toString());
+ brokerProperties.put("AMQPConnections." + connectionName +
".connectionElements.mirror.sync", "false");
+ brokerProperties.put("largeMessageSync", "false");
+ File brokerPropertiesFile = new File(serverLocation,
"broker.properties");
+ saveProperties(brokerProperties, brokerPropertiesFile);
+ }
+
+ @BeforeClass
+ public static void createServers() throws Exception {
+ createServer(DC1_NODE_A, "mirror", DC2_NODEA_URI, 0);
+ createServer(DC2_NODE_A, "mirror", DC1_NODEA_URI, 2);
+ }
+
+ private void startServers() throws Exception {
+ processDC1_node_A = startServer(DC1_NODE_A, -1, -1, new
File(getServerLocation(DC1_NODE_A), "broker.properties"));
+ processDC2_node_A = startServer(DC2_NODE_A, -1, -1, new
File(getServerLocation(DC2_NODE_A), "broker.properties"));
+
+ ServerUtil.waitForServerToStart(0, 10_000);
+ ServerUtil.waitForServerToStart(2, 10_000);
+ }
+
+ @Before
+ public void cleanupServers() {
+ cleanupData(DC1_NODE_A);
+ cleanupData(DC2_NODE_A);
+ }
+
+ private void transactSend(Session session, MessageProducer producer, int
initialCounter, int numberOfMessages) throws Throwable {
+ try {
+ for (int i = initialCounter; i < initialCounter + numberOfMessages;
i++) {
+ TextMessage message;
+ message = session.createTextMessage(largeBody);
+ message.setIntProperty("i", i);
+ message.setBooleanProperty("large", true);
+ String unique = "Unique " + i;
+
message.setStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_DUPLICATE_DETECTION_ID.toString(),
unique);
+ producer.send(message);
+ }
+ session.commit();
+ } catch (JMSException e) {
+ if (e instanceof TransactionRolledBackException &&
e.getMessage().contains("Duplicate message detected")) {
+ logger.debug("OK Exception {}", e.getMessage(), e);
+ return; // ok
+ } else {
+ logger.warn("Not OK Exception {}", e.getMessage(), e);
+ throw e;
+ }
+ }
+ }
+
+
+ @Test
+ public void testAMQP() throws Exception {
+ testACKs("AMQP");
+ }
+
+ @Test
+ public void testCORE() throws Exception {
+ testACKs("CORE");
+ }
+
+ private void testACKs(final String protocol) throws Exception {
+ startServers();
+
+ final int consumers = 10;
+ final int numberOfMessages = 300;
+ final int messagesPerConsumer = numberOfMessages / consumers;
+
+ // Just a reminder: if you change number on this test, this needs to be
true:
+ Assert.assertEquals("Invalid test config", 0, numberOfMessages %
consumers);
+
+ AtomicBoolean running = new AtomicBoolean(true);
+ runAfter(() -> running.set(false));
+
+ String snfQueue = "$ACTIVEMQ_ARTEMIS_MIRROR_mirror";
+
+ ExecutorService executor = Executors.newFixedThreadPool(consumers);
+ runAfter(executor::shutdownNow);
+ ConnectionFactory connectionFactoryDC1A =
CFUtil.createConnectionFactory(protocol, DC1_NODEA_URI);
+ CountDownLatch sendDone = new CountDownLatch(1);
+ CountDownLatch killSend = new CountDownLatch(1);
+
+ executor.execute(() -> {
+ int messagesSent = 0;
+ while (running.get() && messagesSent < numberOfMessages) {
+ try (Connection connection =
connectionFactoryDC1A.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageProducer producer = session.createProducer(queue);
+ if (messagesSent < 100) {
+ transactSend(session, producer, messagesSent, 1);
+ messagesSent++;
+ logger.debug("Sent {}", messagesSent);
+ if (messagesSent == 100) {
+ logger.debug("Signal to kill");
+ killSend.countDown();
+ }
+ } else {
+ transactSend(session, producer, messagesSent, 100);
+ messagesSent += 100;
+ logger.debug("Sent {}", messagesSent);
+ }
+ } catch (Throwable e) {
+ logger.debug(e.getMessage(), e);
+ try {
+ Thread.sleep(100);
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ sendDone.countDown();
+ });
+
+ Assert.assertTrue(killSend.await(10, TimeUnit.SECONDS));
+
+ restartDC1_ServerA();
+
+ Assert.assertTrue(sendDone.await(50, TimeUnit.SECONDS));
+
+ SimpleManagement simpleManagementDC1A = new
SimpleManagement(DC1_NODEA_URI, null, null);
+ SimpleManagement simpleManagementDC2A = new
SimpleManagement(DC2_NODEA_URI, null, null);
+
+ Wait.assertEquals(0, () -> getCount(simpleManagementDC1A, snfQueue));
+ Wait.assertEquals(numberOfMessages, () -> getCount(simpleManagementDC1A,
QUEUE_NAME));
+ Wait.assertEquals(numberOfMessages, () -> getCount(simpleManagementDC2A,
QUEUE_NAME));
+
+ AtomicInteger errors = new AtomicInteger(0);
+
+ CountDownLatch latchKill = new CountDownLatch(consumers);
+
+ CountDownLatch latchDone = new CountDownLatch(consumers);
+
+ Runnable runnableConsumer = () -> {
+ int messagesConsumed = 0;
+ while (running.get() && messagesConsumed < messagesPerConsumer) {
+ try (Connection connection =
connectionFactoryDC1A.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+ while (messagesConsumed < messagesPerConsumer) {
+ Message message = consumer.receive(100);
+ if (message == null || !(message instanceof TextMessage)) {
+ logger.debug("message received={}", message);
+ errors.incrementAndGet();
+ }
+ session.commit();
+ messagesConsumed++;
+ logger.debug("Received {}", messagesConsumed);
+ if (messagesConsumed == 10) {
+ latchKill.countDown();
+ }
+ }
+ } catch (Throwable e) {
+ logger.debug(e.getMessage(), e);
+ try {
+ Thread.sleep(100);
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ latchDone.countDown();
+ };
+
+ for (int i = 0; i < consumers; i++) {
+ executor.execute(runnableConsumer);
+ }
+
+ Assert.assertTrue(latchKill.await(10, TimeUnit.SECONDS));
+
+ restartDC1_ServerA();
+
+ Assert.assertTrue(latchDone.await(1, TimeUnit.MINUTES));
+
+ try (Connection connection = connectionFactoryDC1A.createConnection()) {
+ Session session = connection.createSession(true,
Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue(QUEUE_NAME);
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+ while (true) {
+ if (consumer.receive(500) == null) {
+ break;
+ }
+ }
Review Comment:
the while(true) and if->break could just be collapsed into while receive !=
null
It might also be nice to actually count and then log the received count,
since it might help explain any failure in the following queue count assertion.
Could it even assert on the count here (perhaps after the commit below)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
