AMQ-8525: Try to harden the JMS Client tests

Author: jeanouii

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

@@ -633,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;
+    }
 }