Merge pull request #13719 from rabbitmq/mergify/bp/v4.0.x/pr-13717

Fix test flakes in various testsuites (backport #13470) (backport #13717)

Author: dumbbell

deps/rabbit/test/logging_SUITE.erl

@@ -57,7 +57,7 @@
          logging_to_syslog_works/1]).
 
 suite() ->
-    [{timetrap, {minutes, 1}}].
+    [{timetrap, {minutes, 3}}].
 
 all() ->
     [

deps/rabbit/test/priority_queue_recovery_SUITE.erl

@@ -35,8 +35,10 @@ end_per_suite(Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config).
 
 init_per_group(_, Config) ->
+    Suffix = rabbit_ct_helpers:testcase_absname(Config, "", "-"),
     Config1 = rabbit_ct_helpers:set_config(Config, [
-        {rmq_nodes_count, 2}
+        {rmq_nodes_count, 2},
+        {rmq_nodename_suffix, Suffix}
       ]),
     rabbit_ct_helpers:run_steps(Config1,
       rabbit_ct_broker_helpers:setup_steps() ++

deps/rabbit_common/mk/rabbitmq-run.mk

@@ -320,10 +320,13 @@ start-background-broker: node-tmpdir $(DIST_TARGET)
 	$(BASIC_SCRIPT_ENV_SETTINGS) \
 	  $(RABBITMQ_SERVER) \
 	  $(REDIRECT_STDIO) &
+	trap 'test "$$?" = 0 || $(MAKE) stop-node' EXIT && \
 	ERL_LIBS="$(DIST_ERL_LIBS)" \
 	  $(RABBITMQCTL) -n $(RABBITMQ_NODENAME) wait --timeout $(RMQCTL_WAIT_TIMEOUT) $(RABBITMQ_PID_FILE) && \
-	ERL_LIBS="$(DIST_ERL_LIBS)" \
-	  $(RABBITMQCTL) --node $(RABBITMQ_NODENAME) await_startup
+	for i in $$(seq 1 10); do \
+	  ERL_LIBS="$(DIST_ERL_LIBS)" $(RABBITMQCTL) -n $(RABBITMQ_NODENAME) await_startup || sleep 1; \
+	done && \
+	ERL_LIBS="$(DIST_ERL_LIBS)" $(RABBITMQCTL) -n $(RABBITMQ_NODENAME) await_startup
 
 start-rabbit-on-node:
 	$(exec_verbose) ERL_LIBS="$(DIST_ERL_LIBS)" \

deps/rabbitmq_mqtt/test/v5_SUITE.erl

@@ -206,10 +206,27 @@ end_per_testcase(T, Config) ->
     end_per_testcase0(T, Config).
 
 end_per_testcase0(Testcase, Config) ->
+    %% Terminate all connections and wait for sessions to terminate before
+    %% starting the next test case.
+    _ = rabbit_ct_broker_helpers:rpc(
+          Config, 0,
+          rabbit_networking, close_all_connections, [<<"test finished">>]),
+    _ = rabbit_ct_broker_helpers:rpc_all(
+          Config,
+          rabbit_mqtt, close_local_client_connections, [normal]),
+    eventually(?_assertEqual(
+                  [],
+                  rpc(Config, rabbit_mqtt, local_connection_pids, []))),
     %% Assert that every testcase cleaned up their MQTT sessions.
+    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []),
     eventually(?_assertEqual([], rpc(Config, rabbit_amqqueue, list, []))),
     rabbit_ct_helpers:testcase_finished(Config, Testcase).
 
+delete_queues() ->
+    _ = [catch rabbit_amqqueue:delete(Q, false, false, <<"test finished">>)
+         || Q <- rabbit_amqqueue:list()],
+    ok.
+
 %% -------------------------------------------------------------------
 %% Testsuite cases
 %% -------------------------------------------------------------------

deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/FailureTest.java

@@ -221,6 +221,7 @@ void noLostConfirmedMessagesWhenLeaderGoesAway() throws Exception {
                 () -> {
                   connected.set(false);
 
+                  try { Thread.sleep(2000); } catch (Exception e) {}
                   Client locator =
                       cf.get(new Client.ClientParameters().port(streamPortNode2()));
                   // wait until there's a new leader
@@ -467,6 +468,7 @@ void consumerReattachesToOtherReplicaWhenReplicaGoesAway() throws Exception {
             // avoid long-running task in the IO thread
             executorService.submit(
                 () -> {
+                  try { Thread.sleep(2000); } catch (Exception e) {}
                   Client.StreamMetadata m = metadataClient.metadata(stream).get(stream);
                   int newReplicaPort = m.getReplicas().get(0).getPort();
 

deps/rabbitmq_stream/test/rabbit_stream_SUITE_data/src/test/java/com/rabbitmq/stream/LeaderLocatorTest.java

@@ -28,6 +28,7 @@
 import com.rabbitmq.stream.impl.Client.Response;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import java.util.Collections;
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -57,7 +58,9 @@ void invalidLocatorShouldReturnError() {
   void clientLocalLocatorShouldMakeLeaderOnConnectedNode() {
     int[] ports = new int[] {TestUtils.streamPortNode1(), TestUtils.streamPortNode2()};
     for (int port : ports) {
-      Client client = cf.get(new Client.ClientParameters().port(port));
+      Client client = cf.get(new Client.ClientParameters()
+              .port(port)
+              .rpcTimeout(Duration.ofSeconds(30)));
       String s = UUID.randomUUID().toString();
       try {
         Response response =