[WIP] AMQ-8525: attempt to improve build time

Jenkinsfile

@@ -131,7 +131,7 @@ pipeline {
                 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'
-                sh 'mvn -B -e -fae test -Dsurefire.rerunFailingTestsCount=3'
+                sh 'mvn -B -e -fae test -Dsurefire.rerunFailingTestsCount=3 -Pall-parallel -Dactivemq.tests=parallel'
             }
             post {
                 always {

activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/scheduler/JobSchedulerImpl.java

@@ -268,6 +268,16 @@ private void doReschedule(final String jobId, long executionTime, long nextExecu
         this.store.store(update);
     }
 
+    private void doSchedule(final List<Closure> toSchedule) {
+        for (Closure closure : toSchedule) {
+            try {
+                closure.run();
+            } catch (final Exception e) {
+                LOG.warn("Failed to schedule job", e);
+            }
+        }
+    }
+
     private void doRemove(final List<Closure> toRemove) throws IOException {
         for (Closure closure : toRemove) {
             closure.run();
@@ -727,6 +737,7 @@ protected void mainLoop() {
                 // needed before firing the job event.
                 List<Closure> toRemove = new ArrayList<>();
                 List<Closure> toReschedule = new ArrayList<>();
+                List<Closure> toSchedule = new ArrayList<>();
                 try {
                     this.store.readLockIndex();
 
@@ -776,12 +787,18 @@ protected void mainLoop() {
                                                 // we have a separate schedule to run at this time
                                                 // so the cron job is used to set of a separate schedule
                                                 // hence we won't fire the original cron job to the
-                                                // listeners but we do need to start a separate schedule
-                                                String jobId = ID_GENERATOR.generateId();
-                                                ByteSequence payload = getPayload(job.getLocation());
-                                                schedule(jobId, payload, "", job.getDelay(), job.getPeriod(), job.getRepeat());
-                                                waitTime = job.getDelay() != 0 ? job.getDelay() : job.getPeriod();
-                                                this.scheduleTime.setWaitTime(waitTime);
+                                                // listeners, but we do need to start a separate schedule
+                                                toSchedule.add(() -> {
+                                                    try {
+                                                        String jobId = ID_GENERATOR.generateId();
+                                                        ByteSequence payload = getPayload(job.getLocation());
+                                                        schedule(jobId, payload, "", job.getDelay(), job.getPeriod(), job.getRepeat());
+                                                    } catch (Exception e) {
+                                                        LOG.warn("Failed to schedule cron follow-up job", e);
+                                                    }
+                                                });
+                                                long wait = job.getDelay() != 0 ? job.getDelay() : job.getPeriod();
+                                                this.scheduleTime.setWaitTime(wait);
                                             }
                                         } else {
                                             toRemove.add(() -> doRemove(executionTime, job.getJobId()));
@@ -797,6 +814,10 @@ protected void mainLoop() {
                 } finally {
                     this.store.readUnlockIndex();
 
+                    // deferred execution of all jobs to be scheduled to avoid deadlock with indexLock
+                    doSchedule(toSchedule);
+
+                    // now reschedule all jobs that need rescheduling
                     doReschedule(toReschedule);
 
                     // now remove all jobs that have not been rescheduled,
@@ -805,6 +826,7 @@ protected void mainLoop() {
                 }
 
                 this.scheduleTime.pause();
+
             } catch (Exception ioe) {
                 LOG.error("{} Failed to schedule job", this.name, ioe);
                 try {

activemq-mqtt/pom.xml

@@ -28,6 +28,11 @@
   <artifactId>activemq-mqtt</artifactId>
   <packaging>jar</packaging>
   <name>ActiveMQ :: MQTT Protocol</name>
+
+  <properties>
+    <surefire.version>3.5.3</surefire.version>
+  </properties>
+
   <description>The ActiveMQ MQTT Protocol Implementation</description>
 
   <dependencies>
@@ -198,24 +203,6 @@
       </plugins>
     </pluginManagement>
     <plugins>
-      <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>
-        </configuration>
-      </plugin>
       <plugin>
         <groupId>org.apache.activemq.protobuf</groupId>
         <artifactId>activemq-protobuf</artifactId>
@@ -293,6 +280,83 @@
         </plugins>
       </build>
     </profile>
+    <profile>
+      <id>all-parallel</id>
+      <activation>
+        <activeByDefault>true</activeByDefault>
+        <property>
+          <name>activemq.tests</name>
+          <value>parallel</value>
+        </property>
+      </activation>
+      <properties>
+        <parallel.tests.fork.count>2C</parallel.tests.fork.count>
+      </properties>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-surefire-plugin</artifactId>
+            <version>${surefire.version}</version>
+            <configuration>
+              <skipTests>true</skipTests>
+            </configuration>
+            <executions>
+              <execution>
+                <id>parallel</id>
+                <phase>test</phase>
+                <goals>
+                  <goal>test</goal>
+                </goals>
+                <configuration combine.self="override">
+                  <skipTests>false</skipTests>
+                  <forkCount>${parallel.tests.fork.count}</forkCount>
+                  <reuseForks>false</reuseForks>
+                  <forkedProcessTimeoutInSeconds>600</forkedProcessTimeoutInSeconds>
+                  <runOrder>balanced</runOrder>
+                  <failIfNoTests>false</failIfNoTests>
+                  <groups>org.apache.activemq.transport.mqtt.ParallelTest</groups>
+                  <systemPropertyVariables>
+                    <java.net.preferIPv4Stack>true</java.net.preferIPv4Stack>
+                    <org.apache.activemq.default.directory.prefix>${project.build.directory}/parallel-tests-${surefire.forkNumber}/</org.apache.activemq.default.directory.prefix>
+                    <org.apache.activemq.AutoFailTestSupport.disableSystemExit>true</org.apache.activemq.AutoFailTestSupport.disableSystemExit>
+                    <org.apache.activemq.broker.jmx.createConnector>false</org.apache.activemq.broker.jmx.createConnector>
+                    <!-- 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>
+              <execution>
+                <id>serial</id>
+                <phase>test</phase>
+                <goals>
+                  <goal>test</goal>
+                </goals>
+                <configuration combine.self="override">
+                  <skipTests>false</skipTests>
+                  <runOrder>balanced</runOrder>
+                  <failIfNoTests>false</failIfNoTests>
+                  <excludedGroups>org.apache.activemq.transport.mqtt.ParallelTest</excludedGroups>
+                  <systemPropertyVariables>
+                    <java.net.preferIPv4Stack>true</java.net.preferIPv4Stack>
+                    <org.apache.activemq.default.directory.prefix>${project.build.directory}/</org.apache.activemq.default.directory.prefix>
+                    <org.apache.activemq.AutoFailTestSupport.disableSystemExit>true</org.apache.activemq.AutoFailTestSupport.disableSystemExit>
+                    <org.apache.activemq.broker.jmx.createConnector>false</org.apache.activemq.broker.jmx.createConnector>
+                  </systemPropertyVariables>
+                </configuration>
+              </execution>
+            </executions>
+            <dependencies>
+              <dependency>
+                <groupId>org.apache.maven.surefire</groupId>
+                <artifactId>surefire-junit47</artifactId>
+                <version>${surefire.version}</version>
+              </dependency>
+            </dependencies>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
   </profiles>
 
   <repositories>

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

@@ -39,12 +39,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

@@ -41,10 +41,12 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.junit.experimental.categories.Category;
 /**
  * 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,6 +16,8 @@
  */
 package org.apache.activemq.transport.mqtt;
 
+import org.junit.experimental.categories.Category;
+
 /**
  * Run the basic tests with the NIO Transport.
  */

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/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);