From a30b3a81b9d832938f5f700d17ef6aa38879dc7e Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Fri, 7 May 2021 11:56:22 -0400 Subject: [PATCH] ARTEMIS-3293 System test on Paging and DLQ processing --- tests/smoke-tests/pom.xml | 24 +- .../resources/servers/mmfactory/broker.xml | 244 ++++++++++ .../tests/smoke/common/SmokeTestBase.java | 23 + .../smoke/jmx2/JmxServerControlTest.java | 1 + .../tests/smoke/mmfactory/MMSFactoryTest.java | 439 ++++++++++++++++++ 5 files changed, 730 insertions(+), 1 deletion(-) create mode 100644 tests/smoke-tests/src/main/resources/servers/mmfactory/broker.xml create mode 100644 tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/mmfactory/MMSFactoryTest.java diff --git a/tests/smoke-tests/pom.xml b/tests/smoke-tests/pom.xml index e00dfb3815..ed9c53bcbf 100644 --- a/tests/smoke-tests/pom.xml +++ b/tests/smoke-tests/pom.xml @@ -803,7 +803,29 @@ ${basedir}/target/classes/servers/infinite-redelivery - + + test-compile + create-mmfactory + + create + + + + ${basedir}/target/classes/servers/mmfactory + true + admin + admin + ${basedir}/target/mmfactory + false + + --java-options + -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote=true + -Dcom.sun.management.jmxremote.port=11099 -Dcom.sun.management.jmxremote.rmi.port=11098 + -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false + + + + diff --git a/tests/smoke-tests/src/main/resources/servers/mmfactory/broker.xml b/tests/smoke-tests/src/main/resources/servers/mmfactory/broker.xml new file mode 100644 index 0000000000..b81420fba2 --- /dev/null +++ b/tests/smoke-tests/src/main/resources/servers/mmfactory/broker.xml @@ -0,0 +1,244 @@ + + + + + + + + 0.0.0.0 + + + true + + + NIO + + data/paging + + data/bindings + + data/journal + + data/large-messages + + true + + 2 + + 10 + + 4096 + + 10M + + + 10000 + + true + + + + 1 + + + + + + + + + + + + + + + + + + true + + + 5000 + + + 90 + + + true + + 120000 + + 60000 + + HALT + + + 70560 + + + + + + + + + + + + + + + + + tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true + + + tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpMinLargeMessageSize=102400;amqpDuplicateDetection=true + + + tcp://0.0.0.0:61613?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=STOMP;useEpoll=true + + + tcp://0.0.0.0:5445?anycastPrefix=jms.queue.;multicastPrefix=jms.topic.;protocols=HORNETQ,STOMP;useEpoll=true + + + tcp://0.0.0.0:1883?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=MQTT;useEpoll=true + + + + + + + + + + + + + + + + + + + + + + + + DLQ + ExpiryQueue + 0 + + -1 + 10 + PAGE + true + true + true + true + + + + DLQ + ExpiryQueue + 0 + + 2 + 524680 + 124680 + 10 + PAGE + true + true + true + true + + + + +
+ + + +
+
+ + + +
+ +
+ + +
+
+ + + + true + + +
+ + + + diff --git a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/common/SmokeTestBase.java b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/common/SmokeTestBase.java index 964b749b7f..313b15c862 100644 --- a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/common/SmokeTestBase.java +++ b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/common/SmokeTestBase.java @@ -17,6 +17,9 @@ package org.apache.activemq.artemis.tests.smoke.common; +import javax.management.remote.JMXConnector; +import javax.management.remote.JMXConnectorFactory; +import javax.management.remote.JMXServiceURL; import java.io.File; import java.io.IOException; import java.util.HashSet; @@ -86,4 +89,24 @@ public class SmokeTestBase extends ActiveMQTestBase { return process; } + protected static JMXConnector getJmxConnector(String hostname, int port) throws Exception { + // Without this, the RMI server would bind to the default interface IP (the user's local IP mostly) + System.setProperty("java.rmi.server.hostname", hostname); + + // I don't specify both ports here manually on purpose. See actual RMI registry connection port extraction below. + String urlString = "service:jmx:rmi:///jndi/rmi://" + hostname + ":" + port + "/jmxrmi"; + + JMXServiceURL url = new JMXServiceURL(urlString); + JMXConnector jmxConnector = null; + + try { + jmxConnector = JMXConnectorFactory.connect(url); + System.out.println("Successfully connected to: " + urlString); + } catch (Exception e) { + jmxConnector = null; + e.printStackTrace(); + Assert.fail(e.getMessage()); + } + return jmxConnector; + } } diff --git a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/jmx2/JmxServerControlTest.java b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/jmx2/JmxServerControlTest.java index 421e99c783..a7802e7ea5 100644 --- a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/jmx2/JmxServerControlTest.java +++ b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/jmx2/JmxServerControlTest.java @@ -110,4 +110,5 @@ public class JmxServerControlTest extends SmokeTestBase { jmxConnector.close(); } } + } diff --git a/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/mmfactory/MMSFactoryTest.java b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/mmfactory/MMSFactoryTest.java new file mode 100644 index 0000000000..8756a867e7 --- /dev/null +++ b/tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/mmfactory/MMSFactoryTest.java @@ -0,0 +1,439 @@ +/* + * 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.smoke.mmfactory; + +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.MessageListener; +import javax.jms.MessageProducer; +import javax.jms.Queue; +import javax.jms.Session; +import javax.jms.TextMessage; +import javax.jms.Topic; +import javax.management.MBeanServerConnection; +import javax.management.MBeanServerInvocationHandler; +import javax.management.ObjectName; +import javax.management.remote.JMXConnector; +import java.io.File; +import java.io.FileOutputStream; +import java.io.PrintStream; +import java.util.Arrays; +import java.util.Collection; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration; +import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.api.core.management.ObjectNameBuilder; +import org.apache.activemq.artemis.api.core.management.QueueControl; +import org.apache.activemq.artemis.tests.smoke.common.SmokeTestBase; +import org.apache.activemq.artemis.utils.RandomUtil; +import org.apache.activemq.artemis.utils.SpawnedVMSupport; +import org.apache.activemq.artemis.utils.Wait; +import org.apache.qpid.jms.JmsConnectionFactory; +import org.jboss.logging.Logger; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +@RunWith(Parameterized.class) +public class MMSFactoryTest extends SmokeTestBase { + + private static final Logger logger = Logger.getLogger(MMSFactoryTest.class); + + public static final String SERVER_NAME_0 = "mmfactory"; + private static final String JMX_SERVER_HOSTNAME = "localhost"; + private static final int JMX_SERVER_PORT = 11099; + + final String theprotocol; + final int BATCH_SIZE; + // how many times the server will be restarted + final int restarts; + // how many times the clients will run per restart + final int clientRuns; + // As the produces sends messages, a client will be killed every X messages. This is it! + final int killClientEveryX; + + + @Parameterized.Parameters(name = "protocol={0}, batchSize={1}, restarts={2}, clientRuns={3}, killEveryX={4}") + public static Collection parameters() { + return Arrays.asList(new Object[][]{{"CORE", 2000, 2, 2, 500}, {"AMQP", 2000, 2, 2, 500}}); + } + + public MMSFactoryTest(String protocol, int batchSize, int restarts, int clientRuns, int killClientEveryX) { + this.theprotocol = protocol; + this.BATCH_SIZE = batchSize; + this.restarts = restarts; + this.clientRuns = clientRuns; + this.killClientEveryX = killClientEveryX; + } + + + public static void main(String[] arg) { + + try { + Consumer consumer = new Consumer(arg[0], Integer.parseInt(arg[1]), arg[2], Integer.parseInt(arg[3]), arg[4], Integer.parseInt(arg[5])); + //consumer.run(); + consumer.runListener(); + + while (true) { + Thread.sleep(10_000); + } + } catch (Throwable e) { + System.exit(1); + } + } + + public static String getConsumerLog(int id) { + return getServerLocation(SERVER_NAME_0) + "/data/" + "Consumer" + id + ".log"; + } + + public static ConnectionFactory createConnectionFactory(String protocol, String uri) { + if (protocol.toUpperCase().equals("OPENWIRE")) { + return new org.apache.activemq.ActiveMQConnectionFactory(uri); + } else if (protocol.toUpperCase().equals("AMQP")) { + + if (uri.startsWith("tcp://")) { + // replacing tcp:// by amqp:// + uri = "amqp" + uri.substring(3); + } + return new JmsConnectionFactory(uri); + } else if (protocol.toUpperCase().equals("CORE") || protocol.toUpperCase().equals("ARTEMIS")) { + return new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(uri); + } else { + throw new IllegalStateException("Unkown:" + protocol); + } + } + + Process startConsumerProcess(String protocol, + int slowTime, + String queueName, + int credits, + int consumerID) throws Exception { + + return SpawnedVMSupport.spawnVM(MMSFactoryTest.class.getName(), protocol, "" + slowTime, queueName, "" + credits, getConsumerLog(consumerID), "" + consumerID); + } + + Process serverProcess; + + @Before + public void before() throws Exception { + cleanupData(SERVER_NAME_0); + disableCheckThread(); + serverProcess = startServer(SERVER_NAME_0, 0, 30000); + } + + @Test + public void testMMSorting() throws Exception { + for (int i = 0; i < restarts; i++) { + logger.debug("*******************************************************************************************************************************"); + logger.debug("Starting " + clientRuns); + logger.debug("*******************************************************************************************************************************"); + testMMSorting(clientRuns * i, clientRuns * (i + 1)); + + stopServerWithFile(getServerLocation(SERVER_NAME_0)); + Thread.sleep(1000); + + try { + serverProcess.destroyForcibly(); + } catch (Throwable ignored) { + } + serverProcess = startServer(SERVER_NAME_0, 0, 30000); + } + } + + public void testMMSorting(int countStart, int countEnd) throws Exception { + + JMXConnector jmxConnector = getJmxConnector(JMX_SERVER_HOSTNAME, JMX_SERVER_PORT); + + MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection(); + String brokerName = "0.0.0.0"; // configured e.g. in broker.xml element + ObjectNameBuilder objectNameBuilder = ObjectNameBuilder.create(ActiveMQDefaultConfiguration.getDefaultJmxDomain(), brokerName, true); + //ActiveMQServerControl activeMQServerControl = MBeanServerInvocationHandler.newProxyInstance(mBeanServerConnection, objectNameBuilder.getActiveMQServerObjectName(), ActiveMQServerControl.class, false); + + ObjectName queueObjectName = objectNameBuilder.getQueueObjectName(SimpleString.toSimpleString("MMFactory"), SimpleString.toSimpleString("MMConsumer"), RoutingType.MULTICAST); + QueueControl queueControl = MBeanServerInvocationHandler.newProxyInstance(mBeanServerConnection, queueObjectName, QueueControl.class, false); + + final int NUMBER_OF_CONSUMERS = 6; + final int NUMBER_OF_FASTCONSUMERS = 0; // not using at the moment + + Process[] consumers = new Process[NUMBER_OF_CONSUMERS + NUMBER_OF_FASTCONSUMERS]; + int[] timeForConsumers = new int[NUMBER_OF_CONSUMERS + NUMBER_OF_FASTCONSUMERS]; + + for (int i = 0; i < NUMBER_OF_CONSUMERS; i++) { + timeForConsumers[i] = (i % 2 == 0 ? 200 : 300); + } + + timeForConsumers[1] = 100; + timeForConsumers[5] = 500; + + for (int i = NUMBER_OF_CONSUMERS; i < NUMBER_OF_CONSUMERS + NUMBER_OF_FASTCONSUMERS; i++) { + timeForConsumers[i] = 0; + } + + for (int i = 0; i < consumers.length; i++) { + consumers[i] = startConsumerProcess(theprotocol, timeForConsumers[i], "MMFactory::MMConsumer", 100, i); + } + + Process dlqProcess = startConsumerProcess(theprotocol, 0, "DLQ", 100, 1000); + + AtomicInteger retryNumber = new AtomicInteger(0); + int expectedTotalSize = 0; + + ConnectionFactory factory = createConnectionFactory(theprotocol, "tcp://localhost:61616"); + try (Connection connection = factory.createConnection()) { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Topic queue = session.createTopic("MMFactory"); + MessageProducer mmsFactory = session.createProducer(queue); + + Topic controlTopic = session.createTopic("MMControl"); + + + String largeString; + { + StringBuffer largeStringBuffer = new StringBuffer(); + + while (largeStringBuffer.length() < 10) { + largeStringBuffer.append(RandomUtil.randomString()); + } + largeString = largeStringBuffer.toString(); + } + + try { + for (int run = countStart; run <= countEnd; run++) { + AtomicInteger lastTime = new AtomicInteger((int)queueControl.getMessagesAcknowledged()); + for (int i = 0; i < BATCH_SIZE; i++) { + if (i > 0 && i % killClientEveryX == 0) { + System.out.println("REconnecting..."); + logger.debug("Reconnecting..."); + consumers[0].destroyForcibly(); + consumers[0] = startConsumerProcess(theprotocol, timeForConsumers[0], "MMFactory::MMConsumer", 100, 0); + consumers[1].destroyForcibly(); + consumers[1] = startConsumerProcess(theprotocol, timeForConsumers[1], "MMFactory::MMConsumer", 100, 1); + logger.debug("...Reconnected"); + logger.debug("retry=" + retryNumber + ",sent=" + i + ", acked on this batch = " + (queueControl.getMessagesAcknowledged() - (retryNumber.get() * BATCH_SIZE * 2)) + ", total acked = " + queueControl.getMessagesAcknowledged()); + } + TextMessage message = session.createTextMessage("This is blue " + largeString); + message.setStringProperty("color", "blue"); + message.setIntProperty("i", i); + mmsFactory.send(message); + + message = session.createTextMessage("This is red " + largeString); + message.setStringProperty("color", "red"); + message.setIntProperty("i", i); + mmsFactory.send(message); + + if (i % 10 == 0) { + logger.debug("Sending " + i + " run = " + run); + } + + if (i == 0) { + waitForAckChange(queueControl, lastTime); + } + } + + Session sessionControl = connection.createSession(true, Session.SESSION_TRANSACTED); + MessageProducer producerControl = sessionControl.createProducer(controlTopic); + Message controlmessage = sessionControl.createMessage(); + controlmessage.setStringProperty("control", "flush"); + producerControl.send(controlmessage); + sessionControl.commit(); + sessionControl.close(); + + Wait.assertTrue(() -> { + // We will wait a bit here until it's a least a bit closer to the whole Batch + if ((queueControl.getMessagesAcknowledged() + queueControl.getMessagesKilled() + queueControl.getMessagesExpired()) - (retryNumber.get() * BATCH_SIZE * 2) > (BATCH_SIZE * 2 - 500)) { + return true; + } else { + logger.debug("Received " + queueControl.getMessagesAcknowledged()); + return false; + } + }, 45_000, 1_000); + + expectedTotalSize += BATCH_SIZE * 2; + Wait.assertEquals(expectedTotalSize, queueControl::getMessagesAdded); + + Wait.assertEquals(expectedTotalSize, () -> queueControl.getMessagesAcknowledged() + queueControl.getMessagesKilled()); + retryNumber.incrementAndGet(); + + for (Process c : consumers) { + c.destroyForcibly(); + } + for (int i = 0; i < consumers.length; i++) { + File file = new File(getConsumerLog(i)); + if (!file.delete()) { + logger.debug("not possible to remove " + file); + } + } + for (int r = 0; r < consumers.length; r++) { + consumers[r] = startConsumerProcess(theprotocol, timeForConsumers[r], "MMFactory::MMConsumer", 100, r); + } + } + + Thread.sleep(1000); + } finally { + for (Process c : consumers) { + c.destroyForcibly(); + } + + dlqProcess.destroyForcibly(); + + for (int i = 0; i < consumers.length; i++) { + File file = new File(getConsumerLog(i)); + if (!file.delete()) { + logger.warn("not possible to remove " + file); + } + } + + File file = new File(getConsumerLog(1000)); //the DLQ processing ID used + if (!file.delete()) { + logger.warn("not possible to remove " + file); + } + } + } + + } + + private void waitForAckChange(QueueControl queueControl, AtomicInteger lastTime) throws Exception { + Wait.waitFor(() -> { + + if (lastTime.get() == queueControl.getMessagesAcknowledged()) { + logger.debug("Waiting some change on " + queueControl.getMessagesAcknowledged() + " with messages Added = " + queueControl.getMessagesAdded() + " and killed = " + queueControl.getMessagesKilled()); + return false; + } else { + logger.debug("Condition met! with " + queueControl.getMessagesAcknowledged() + " with messages Added = " + queueControl.getMessagesAdded() + " and killed = " + queueControl.getMessagesKilled()); + lastTime.set((int)queueControl.getMessagesAcknowledged()); + return true; + } + }, 5_000); + } + + static class Consumer implements MessageListener { + + boolean clientAck = false; + volatile int slowTime; + final String queuename; + final int credits; + final String protocol; + final int id; + ConnectionFactory factory; + Connection connection; + Session session; + MessageConsumer consumer; + MessageConsumer controlConsumer; + Queue queue; + + Session sessionControl; + + PrintStream fileStream; + + Consumer(String protocol, int slowTime, String queueName, int credits, String logOutput, int id) throws Exception { + this.slowTime = slowTime; + this.queuename = queueName; + this.credits = credits; + this.protocol = protocol; + fileStream = new PrintStream(new FileOutputStream(logOutput, true), true); + this.id = id; + } + + @Override + public void onMessage(Message message) { + try { + String color = message.getStringProperty("color"); + int messageSequence = message.getIntProperty("i"); + if (queuename.equals("DLQ")) { + logger.debug("Processing DLQ on color=" + color + ", sequence=" + messageSequence); + } else if (slowTime > 0) { + Thread.sleep(slowTime); + } + + if (clientAck) { + message.acknowledge(); + } + + fileStream.println(color + ";" + messageSequence); + } catch (Throwable e) { + e.printStackTrace(); + System.exit(-1); + } + + } + + class ControlListener implements MessageListener { + + @Override + public void onMessage(Message message) { + try { + + // This is received at the client, on a remote machine, System.out is the best option to log here + System.out.println("Received control message"); + if (message.getStringProperty("control").equals("flush")) { + Consumer.this.slowTime = 0; + System.out.println("Setting slow time to 0"); + } + sessionControl.commit(); + } catch (Throwable e) { + e.printStackTrace(); + } + } + } + + public void runListener() { + + //factory = createConnectionFactory(protocol, "tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=" + credits); + factory = createConnectionFactory(protocol, "tcp://localhost:61616"); + + System.out.println("Starting"); + connect(); + try { + consumer.setMessageListener(Consumer.this); + } catch (Exception e) { + e.printStackTrace(); + System.exit(-1); + } + } + + private void connect() { + try { + if (connection != null) { + connection.close(); + } + + connection = factory.createConnection(); + connection.start(); + session = connection.createSession(false, clientAck ? Session.CLIENT_ACKNOWLEDGE : Session.AUTO_ACKNOWLEDGE); + queue = session.createQueue(queuename); + consumer = session.createConsumer(queue); + + sessionControl = connection.createSession(true, Session.SESSION_TRANSACTED); + Topic topic = sessionControl.createTopic("MMControl"); + controlConsumer = sessionControl.createSharedDurableConsumer(topic, "consumer" + id); + controlConsumer.setMessageListener(new ControlListener()); + + + } catch (Exception e) { + e.printStackTrace(); + System.exit(-1); + } + } + } + +}