Merge branch 'main' into AMQ-9824

Author: cshannon

activemq-amqp/pom.xml

@@ -180,6 +180,80 @@
     </dependency>
   </dependencies>
 
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <version>${surefire.version}</version>
+        <configuration>
+          <argLine>${surefire.argLine}</argLine>
+          <runOrder>alphabetical</runOrder>
+          <reportFormat>plain</reportFormat>
+          <failIfNoTests>false</failIfNoTests>
+          <excludedGroups>org.apache.activemq.transport.amqp.ParallelTest</excludedGroups>
+          <systemPropertyVariables>
+            <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>
+          <includes>
+            <include>**/*Test.*</include>
+          </includes>
+          <excludes>
+            <exclude>**/*LoadTest.java</exclude>
+            <exclude>**/*StressTest.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.amqp.ParallelTest</groups>
+              <forkCount>2C</forkCount>
+              <reuseForks>false</reuseForks>
+              <forkedProcessTimeoutInSeconds>600</forkedProcessTimeoutInSeconds>
+              <failIfNoTests>false</failIfNoTests>
+              <systemPropertyVariables>
+                <org.apache.activemq.default.directory.prefix>${project.build.directory}/parallel-tests-${surefire.forkNumber}/</org.apache.activemq.default.directory.prefix>
+                <!-- when running tests in parallel in the CI (quite slow) we need to bump the wireformat negotiation timeout (5s by default) -->
+                <org.apache.activemq.transport.wireFormatNegotiationTimeout>30000</org.apache.activemq.transport.wireFormatNegotiationTimeout>
+              </systemPropertyVariables>
+            </configuration>
+          </execution>
+        </executions>
+        <dependencies>
+          <dependency>
+            <groupId>org.apache.maven.surefire</groupId>
+            <artifactId>surefire-junit47</artifactId>
+            <version>${surefire.version}</version>
+          </dependency>
+          <dependency>
+            <groupId>me.fabriciorby</groupId>
+            <artifactId>maven-surefire-junit5-tree-reporter</artifactId>
+            <version>1.5.1</version>
+          </dependency>
+        </dependencies>
+      </plugin>
+    </plugins>
+  </build>
+
   <profiles>
     <profile>
       <!-- profile which is activated is the swiftmq-client-home prop is defined.

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpAndMqttTest.java

@@ -39,7 +39,9 @@
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
+@Category(ParallelTest.class)
 public class AmqpAndMqttTest {
 
     protected BrokerService broker;

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpAndStompInteropTest.java

@@ -40,10 +40,12 @@
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.junit.rules.TestName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Category(ParallelTest.class)
 public class AmqpAndStompInteropTest {
 
     private static final Logger LOG = LoggerFactory.getLogger(AmqpAndStompInteropTest.class);

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java

@@ -51,6 +51,7 @@
 import org.apache.activemq.spring.SpringSslContext;
 import org.apache.activemq.store.kahadb.KahaDBStore;
 import org.apache.activemq.transport.amqp.protocol.AmqpConnection;
+import org.apache.activemq.util.IOHelper;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -61,7 +62,6 @@
 public class AmqpTestSupport {
 
     public static final String MESSAGE_NUMBER = "MessageNumber";
-    public static final String KAHADB_DIRECTORY = "target/activemq-data/";
 
     @Rule public TestName name = new TestName();
 
@@ -120,7 +120,7 @@ protected void createBroker(boolean deleteAllMessages) throws Exception {
         brokerService.setDeleteAllMessagesOnStartup(deleteAllMessages);
         if (isPersistent()) {
             KahaDBStore kaha = new KahaDBStore();
-            kaha.setDirectory(new File(KAHADB_DIRECTORY + getTestName()));
+            kaha.setDirectory(new File(IOHelper.getDefaultDataDirectory() + getTestName()));
             brokerService.setPersistenceAdapter(kaha);
         }
         brokerService.setSchedulerSupport(isSchedulerEnabled());

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTransformerTest.java

@@ -45,9 +45,11 @@
 import org.apache.activemq.transport.amqp.client.AmqpSession;
 import org.junit.After;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Category(ParallelTest.class)
 public class AmqpTransformerTest {
 
     private static final Logger LOG = LoggerFactory.getLogger(AmqpTransformerTest.class);

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientSimpleAuthTest.java

@@ -39,10 +39,12 @@
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.junit.rules.TestName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Category(ParallelTest.class)
 public class JMSClientSimpleAuthTest {
 
     @Rule public TestName name = new TestName();

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java

@@ -61,10 +61,12 @@
 import org.apache.activemq.util.Wait;
 import org.apache.qpid.jms.JmsConnectionFactory;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 import org.objectweb.jtests.jms.framework.TestConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Category(ParallelTest.class)
 public class JMSClientTest extends JMSClientTestSupport {
 
     protected static final Logger LOG = LoggerFactory.getLogger(JMSClientTest.class);
@@ -631,7 +633,7 @@ public void testBrokerRestartWontHangConnectionClose() throws Exception {
     @Test(timeout=30 * 1000)
     public void testProduceAndConsumeLargeNumbersOfMessages() throws Exception {
         int count = 1000;
-        connection = createConnection();
+        connection = createConnectionWithRetry(name.toString(), false);
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Queue queue = session.createQueue(getDestinationName());
         connection.start();

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTestSupport.java

@@ -145,4 +145,37 @@ protected Connection createConnection(String clientId, boolean syncPublish) thro
         connection.start();
         return connection;
     }
+
+    /**
+     * Some SSL/NIO test combinations can occasionally fail the initial frame negotiation
+     * when brokers are started in parallel forks. Retry once on the specific framing error
+     * to smooth out that transient race without hiding other issues.
+     */
+    protected Connection createConnectionWithRetry(String clientId, boolean syncPublish) throws JMSException, InterruptedException {
+        JMSException lastException = null;
+        for (int attempt = 0; attempt < 2; attempt++) {
+            try {
+                return createConnection(clientId, syncPublish);
+            } catch (JMSException ex) {
+                lastException = ex;
+                if (!containsAmqpHeaderMismatch(ex)) {
+                    throw ex;
+                }
+                LOG.warn("AMQP header mismatch on connection attempt {} for {}, retrying once", attempt + 1, getBrokerURI());
+                Thread.sleep(500);
+            }
+        }
+        throw lastException;
+    }
+
+    private boolean containsAmqpHeaderMismatch(Throwable throwable) {
+        Throwable current = throwable;
+        while (current != null) {
+            if (current.getMessage() != null && current.getMessage().contains("AMQP header mismatch")) {
+                return true;
+            }
+            current = current.getCause();
+        }
+        return false;
+    }
 }

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/ParallelTest.java

@@ -0,0 +1,27 @@
+/**
+ * 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.transport.amqp;
+
+
+/**
+ * Marker interface used with {@code @Category(ParallelTest.class)} to opt a
+ * test class or method into the {@code all-parallel} Maven profile. Only tests
+ * explicitly tagged with this category execute when the profile is enabled,
+ * which allows a gradual migration toward full parallelism.
+ */
+public interface ParallelTest {
+}

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpAnonymousSenderTest.java

@@ -28,11 +28,14 @@
 import org.apache.activemq.transport.amqp.client.AmqpReceiver;
 import org.apache.activemq.transport.amqp.client.AmqpSender;
 import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.apache.activemq.transport.amqp.ParallelTest;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
 /**
  * Test for support of Anonymous sender links.
  */
+@Category(ParallelTest.class)
 public class AmqpAnonymousSenderTest extends AmqpClientTestSupport {
 
     @Test(timeout = 60000)