#AMQ-8525 Improve performance of mqtt module

Jenkinsfile

@@ -45,7 +45,6 @@ pipeline {
         choice(name: 'nodeLabel', choices: ['ubuntu', 's390x', 'arm', 'Windows']) 
         choice(name: 'jdkVersion', choices: ['jdk_17_latest', 'jdk_21_latest', 'jdk_24_latest', 'jdk_17_latest_windows', 'jdk_21_latest_windows', 'jdk_24_latest_windows'])
         booleanParam(name: 'deployEnabled', defaultValue: false)
-        booleanParam(name: 'parallelTestsEnabled', defaultValue: true)
         booleanParam(name: 'sonarEnabled', defaultValue: false)
         booleanParam(name: 'testsEnabled', defaultValue: true)
     }
@@ -127,20 +126,12 @@ pipeline {
             }
             when { expression { return params.testsEnabled } }
             steps {
+                echo 'Running tests'
                 sh 'java -version'
                 sh 'mvn -version'
-
                 // all tests is very very long (10 hours on Apache Jenkins)
                 // sh 'mvn -B -e test -pl activemq-unit-tests -Dactivemq.tests=all'
-                script {
-                    if (params.parallelTestsEnabled == 'true') {
-                        sh 'echo "Running parallel-tests ..."'
-                        sh 'mvn -B -e -fae -Pparallel-tests test -Dsurefire.rerunFailingTestsCount=3'
-                    } else {
-                        sh 'echo "Running tests ..."'
-                        sh 'mvn -B -e -fae test -Dsurefire.rerunFailingTestsCount=3'
-                    }
-                }
+                sh 'mvn -B -e -fae test -Dsurefire.rerunFailingTestsCount=3'
             }
             post {
                 always {

activemq-mqtt/pom.xml

@@ -28,6 +28,7 @@
   <artifactId>activemq-mqtt</artifactId>
   <packaging>jar</packaging>
   <name>ActiveMQ :: MQTT Protocol</name>
+
   <description>The ActiveMQ MQTT Protocol Implementation</description>
 
   <dependencies>
@@ -201,21 +202,56 @@
       <plugin>
         <artifactId>maven-surefire-plugin</artifactId>
         <configuration>
-          <forkCount>1</forkCount>
-          <reuseForks>false</reuseForks>
           <argLine>${surefire.argLine}</argLine>
           <runOrder>alphabetical</runOrder>
-           <systemPropertyValues>
-              <org.apache.activemq.default.directory.prefix>target</org.apache.activemq.default.directory.prefix>
-           </systemPropertyValues>
-           <!-- includes>
-             <include>**/*Test.*</include>
-           </includes -->
-           <excludes>
-             <exclude>**/PahoMQTNioTTest.java</exclude>
-           </excludes>
+          <reportFormat>plain</reportFormat>
+          <excludedGroups>org.apache.activemq.transport.mqtt.ParallelTest</excludedGroups>
+          <systemPropertyVariables>
+            <java.net.preferIPv4Stack>true</java.net.preferIPv4Stack>
+            <org.apache.activemq.AutoFailTestSupport.disableSystemExit>true</org.apache.activemq.AutoFailTestSupport.disableSystemExit>
+            <org.apache.activemq.broker.jmx.createConnector>false</org.apache.activemq.broker.jmx.createConnector>
+            <org.apache.activemq.default.directory.prefix>${project.build.directory}/</org.apache.activemq.default.directory.prefix>
+          </systemPropertyVariables>
+          <consoleOutputReporter>
+            <disable>true</disable>
+          </consoleOutputReporter>
+          <statelessTestsetInfoReporter
+            implementation="org.apache.maven.plugin.surefire.extensions.junit5.JUnit5StatelessTestsetInfoTreeReporter">
+            <printStacktraceOnError>true</printStacktraceOnError>
+            <printStacktraceOnFailure>true</printStacktraceOnFailure>
+            <printStdoutOnError>true</printStdoutOnError>
+            <printStdoutOnFailure>true</printStdoutOnFailure>
+            <printStderrOnError>true</printStderrOnError>
+            <printStderrOnFailure>true</printStderrOnFailure>
+          </statelessTestsetInfoReporter>
+          <excludes>
+            <exclude>**/PahoMQTNioTTest.java</exclude>
+          </excludes>
         </configuration>
+        <executions>
+          <execution>
+            <id>parallel</id>
+            <phase>test</phase>
+            <goals>
+              <goal>test</goal>
+            </goals>
+            <configuration>
+              <!-- drop the default excludedGroups -->
+              <excludedGroups combine.self="override"/>
+              <groups>org.apache.activemq.transport.mqtt.ParallelTest</groups>
+              <forkCount>2C</forkCount>
+              <reuseForks>false</reuseForks>
+              <forkedProcessTimeoutInSeconds>600</forkedProcessTimeoutInSeconds>
+              <systemPropertyVariables>
+                <org.apache.activemq.default.directory.prefix>${project.build.directory}/parallel-tests-${surefire.forkNumber}/</org.apache.activemq.default.directory.prefix>
+                <!-- when running MQTT tests in parallel in the CI (quite slow) we need to bump the wireformat negotiation timeout (5s by default) -->
+                <org.apache.activemq.transport.wireFormatNegotiationTimeout>20000</org.apache.activemq.transport.wireFormatNegotiationTimeout>
+              </systemPropertyVariables>
+            </configuration>
+          </execution>
+        </executions>
       </plugin>
+
       <plugin>
         <groupId>org.apache.activemq.protobuf</groupId>
         <artifactId>activemq-protobuf</artifactId>

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTAuthTest.java

@@ -47,15 +47,16 @@
 import org.fusesource.mqtt.codec.CONNACK;
 import org.fusesource.mqtt.codec.MQTTFrame;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 /**
  * Tests various use cases that require authentication or authorization over MQTT
  */
+@Category(ParallelTest.class)
 @RunWith(Parameterized.class)
 public class MQTTAuthTest extends MQTTAuthTestSupport {
 

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCodecTest.java

@@ -41,12 +41,14 @@
 import org.fusesource.mqtt.codec.UNSUBSCRIBE;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * Tests the functionality of the MQTTCodec class.
  */
+@Category(ParallelTest.class)
 public class MQTTCodecTest {
 
     private static final Logger LOG = LoggerFactory.getLogger(MQTTCodecTest.class);

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCompositeQueueRetainedTest.java

@@ -42,9 +42,12 @@
 import org.apache.activemq.util.ByteSequence;
 import org.junit.Test;
 
+import org.junit.experimental.categories.Category;
+
 /**
  *
  */
+@Category(ParallelTest.class)
 public class MQTTCompositeQueueRetainedTest extends MQTTTestSupport {
 
     // configure composite topic

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTConnectTest.java

@@ -35,6 +35,7 @@
 import org.junit.After;
 import org.junit.Ignore;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
@@ -45,6 +46,7 @@
  * Test that connection attempts that don't send a CONNECT frame will
  * get cleaned up by the inactivity monitor.
  */
+@Category(ParallelTest.class)
 @RunWith(Parameterized.class)
 public class MQTTConnectTest extends MQTTTestSupport {
 

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTMaxFrameSizeTest.java

@@ -31,9 +31,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.junit.experimental.categories.Category;
 /**
  * Test that the maxFrameSize configuration value is applied across the transports.
  */
+@Category(ParallelTest.class)
 @RunWith(Parameterized.class)
 public class MQTTMaxFrameSizeTest extends MQTTTestSupport {
 

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTNIOSSLTest.java

@@ -16,9 +16,12 @@
  */
 package org.apache.activemq.transport.mqtt;
 
+import org.junit.experimental.categories.Category;
+
 /**
  * Run the basic tests with the NIO Transport.
  */
+@Category(ParallelTest.class)
 public class MQTTNIOSSLTest extends MQTTTest {
 
     @Override

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTNIOTest.java

@@ -16,9 +16,12 @@
  */
 package org.apache.activemq.transport.mqtt;
 
+import org.junit.experimental.categories.Category;
+
 /**
  * Run the basic tests with the NIO Transport.
  */
+@Category(ParallelTest.class)
 public class MQTTNIOTest extends MQTTTest {
 
     @Override

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTOverlapedSubscriptionsTest.java

@@ -29,7 +29,9 @@
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
+@Category(ParallelTest.class)
 public class MQTTOverlapedSubscriptionsTest {
 
     private BrokerService brokerService;

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTPingReqTest.java

@@ -52,10 +52,12 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.junit.experimental.categories.Category;
 /**
  * Test to show that a PINGRESP will only be sent for a PINGREQ
  * packet after a CONNECT packet has been received.
  */
+@Category(ParallelTest.class)
 @RunWith(Parameterized.class)
 public class MQTTPingReqTest extends MQTTTestSupport {
 

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverterTest.java

@@ -37,12 +37,14 @@
 import org.fusesource.mqtt.codec.MQTTFrame;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 
 /**
  * Tests for various usage scenarios of the protocol converter
  */
+@Category(ParallelTest.class)
 public class MQTTProtocolConverterTest {
 
     private MQTTTransport transport;

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTSSLTest.java

@@ -16,9 +16,12 @@
  */
 package org.apache.activemq.transport.mqtt;
 
+import org.junit.experimental.categories.Category;
+
 /**
  * Run the basic tests with the NIO Transport.
  */
+@Category(ParallelTest.class)
 public class MQTTSSLTest extends MQTTTest {
 
     @Override

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTSubscriptionRecoveryTest.java

@@ -34,9 +34,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.junit.experimental.categories.Category;
 /**
  * Test that all previous QoS 2 subscriptions are recovered on Broker restart.
  */
+@Category(ParallelTest.class)
 @RunWith(Parameterized.class)
 public class MQTTSubscriptionRecoveryTest extends MQTTTestSupport {
 

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java

@@ -65,9 +65,13 @@
 import org.fusesource.mqtt.codec.MQTTFrame;
 import org.fusesource.mqtt.codec.PUBLISH;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Category(ParallelTest.class)
 public class MQTTTest extends MQTTTestSupport {
 
     private static final Logger LOG = LoggerFactory.getLogger(MQTTTest.class);

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTestSupport.java

@@ -39,6 +39,7 @@
 import org.apache.activemq.broker.jmx.TopicViewMBean;
 import org.apache.activemq.store.kahadb.KahaDBStore;
 import org.apache.activemq.transport.mqtt.util.ResourceLoadingSslContext;
+import org.apache.activemq.util.IOHelper;
 import org.fusesource.mqtt.client.MQTT;
 import org.fusesource.mqtt.client.Tracer;
 import org.fusesource.mqtt.codec.MQTTFrame;
@@ -53,8 +54,6 @@ public class MQTTTestSupport {
 
     private static final Logger LOG = LoggerFactory.getLogger(MQTTTestSupport.class);
 
-    public static final String KAHADB_DIRECTORY = "target/activemq-data/";
-
     protected BrokerService brokerService;
     protected int port;
     protected String jmsUri = "vm://localhost";
@@ -143,7 +142,7 @@ protected BrokerService createBroker(boolean deleteAllMessages) throws Exception
         brokerService.setPersistent(isPersistent());
         if (isPersistent()) {
             KahaDBStore kaha = new KahaDBStore();
-            kaha.setDirectory(new File(KAHADB_DIRECTORY + getTestName()));
+            kaha.setDirectory(new File(IOHelper.getDefaultDataDirectory() + "/" + getTestName()));
             brokerService.setPersistenceAdapter(kaha);
         }
         brokerService.setAdvisorySupport(advisorySupport);

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTVirtualTopicSubscriptionsTest.java

@@ -42,6 +42,7 @@
 import org.fusesource.mqtt.client.Topic;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,6 +56,7 @@
 /**
  * Run the basic tests with the NIO Transport.
  */
+@Category(ParallelTest.class)
 public class MQTTVirtualTopicSubscriptionsTest extends MQTTTest {
     private static final Logger LOG = LoggerFactory.getLogger(MQTTVirtualTopicSubscriptionsTest.class);
 

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTWillTest.java

@@ -18,12 +18,14 @@
 
 import org.fusesource.mqtt.client.*;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 
+@Category(ParallelTest.class)
 public class MQTTWillTest extends MQTTTestSupport {
 
     @Test(timeout = 60 * 1000)

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/PahoMQTTNIOTest.java

@@ -16,9 +16,12 @@
  */
 package org.apache.activemq.transport.mqtt;
 
+import org.junit.experimental.categories.Category;
+
 /**
  * Test the NIO transport with this Test group
  */
+@Category(ParallelTest.class)
 public class PahoMQTTNIOTest extends PahoMQTTTest {
 
     @Override

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/PahoMQTTTest.java

@@ -47,6 +47,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.junit.experimental.categories.Category;
+
+@Category(ParallelTest.class)
 public class PahoMQTTTest extends MQTTTestSupport {
 
     private static final Logger LOG = LoggerFactory.getLogger(PahoMQTTTest.class);

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/PahoVirtualTopicMQTTTest.java

@@ -22,12 +22,14 @@
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
 import jakarta.jms.MessageConsumer;
 import jakarta.jms.Session;
 
 import static org.junit.Assert.assertEquals;
 
+@Category(ParallelTest.class)
 public class PahoVirtualTopicMQTTTest extends PahoMQTTTest {
 
     @Override