Improve robustness of the AMQP test

integration-tests/reactive-messaging-amqp/src/test/java/io/quarkus/it/amqp/AmqpBroker.java

@@ -1,5 +1,7 @@
 package io.quarkus.it.amqp;
 
+import static org.awaitility.Awaitility.await;
+
 import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
@@ -13,9 +15,11 @@
 
 public class AmqpBroker implements QuarkusTestResourceLifecycleManager {
 
-    private EmbeddedActiveMQ server;
+    static EmbeddedActiveMQ server;
+    static AmqpBroker instance;
 
     public Map<String, String> start() {
+        instance = this;
         try {
             server = new EmbeddedActiveMQ();
             server.setSecurityManager(new ActiveMQSecurityManager() {
@@ -34,6 +38,10 @@ public boolean validateUserAndRole(String username, String password, Set<Role> s
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
+
+        await().until(() -> server.getActiveMQServer() != null
+                && server.getActiveMQServer().isActive()
+                && server.getActiveMQServer().getConnectorsService().isStarted());
         return Collections.emptyMap();
     }
 
@@ -42,9 +50,17 @@ public void stop() {
             if (server != null) {
                 server.stop();
             }
+            instance = null;
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
     }
 
+    public static void restartBroker() {
+        if (instance != null) {
+            instance.stop();
+        }
+        new AmqpBroker().start();
+    }
+
 }

integration-tests/reactive-messaging-amqp/src/test/java/io/quarkus/it/amqp/AmqpConnectorTest.java

@@ -5,6 +5,8 @@
 
 import java.util.List;
 
+import org.awaitility.core.ConditionTimeoutException;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.DisabledOnOs;
 import org.junit.jupiter.api.condition.OS;
@@ -20,10 +22,28 @@ public class AmqpConnectorTest {
 
     protected static final TypeRef<List<Person>> TYPE_REF = new TypeRef<List<Person>>() {
     };
+    protected static final int RETRY_ATTEMPTS = 10;
 
     @Test
     public void test() {
-        await().until(() -> get("/amqp/people").as(TYPE_REF).size() >= 6);
+        for (int i = 0; i < RETRY_ATTEMPTS; i++) {
+            try {
+                await()
+                        .until(() -> get("/amqp/people").as(TYPE_REF).size() >= 6);
+                return;
+            } catch (ConditionTimeoutException e) {
+                // Sometimes the AMQP broker is just in a broken state, in this case, restart it.
+                restartBroker();
+            } catch (Exception e) {
+                Assertions.fail(e);
+                return;
+            }
+        }
+        Assertions.fail("Unable to run the AMQP test successfully, despite " + RETRY_ATTEMPTS + " restarts of the broker");
+    }
+
+    private void restartBroker() {
+        AmqpBroker.restartBroker();
     }
 
 }