From e7f676d5cb720ae3ef71892efd8e3d68eca58e17 Mon Sep 17 00:00:00 2001
From: Karl Nilsson <kjnilsson@gmail.com>
Date: Wed, 6 Mar 2024 09:46:33 +0000
Subject: [PATCH] rabbit_stream_queue_SUITE: test reliability with khepri

Khepri projections may not be immedately updated after a queue
declare so we need to await_condition to validate the leader
pid is updated correctly.
---
 .../rabbit/test/rabbit_stream_queue_SUITE.erl | 26 +++++++++++++------
 1 file changed, 18 insertions(+), 8 deletions(-)

diff --git a/deps/rabbit/test/rabbit_stream_queue_SUITE.erl b/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
index b50fb4537307..ae47b025e071 100644
--- a/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
+++ b/deps/rabbit/test/rabbit_stream_queue_SUITE.erl
@@ -16,6 +16,7 @@
 -compile(nowarn_export_all).
 -compile(export_all).
 
+-import(rabbit_ct_helpers, [await_condition/2]).
 -define(WAIT, 5000).
 
 suite() ->
@@ -2168,8 +2169,10 @@ leader_locator_client_local(Config) ->
                  declare(Config, Server1, Q, [{<<"x-queue-type">>, longstr, <<"stream">>},
                                               {<<"x-queue-leader-locator">>, longstr, <<"client-local">>}])),
 
-    ?assertMatch(Server1, proplists:get_value(leader,
-                                              find_queue_info(Config, [leader]))),
+    await_condition(
+      fun () ->
+              Server1 == proplists:get_value(leader, find_queue_info(Config, [leader]))
+      end, 60),
 
     ?assertMatch(#'queue.delete_ok'{},
                  delete(Config, Server1, Q)),
@@ -2180,8 +2183,13 @@ leader_locator_client_local(Config) ->
                  declare(Config, Server2, Q2, [{<<"x-queue-type">>, longstr, <<"stream">>},
                                               {<<"x-queue-leader-locator">>, longstr, <<"client-local">>}])),
 
-    ?assertMatch(Server2, proplists:get_value(leader,
-                                              find_queue_info(Q2, Config, 0, [leader]))),
+    %% the amqqueue:pid field is updated async for khepri
+    %% so we need to await the condition here
+    await_condition(
+      fun () ->
+              Server2 == proplists:get_value(leader,
+                                             find_queue_info(Q2, Config, 0, [leader]))
+      end, 60),
 
     ?assertMatch(#'queue.delete_ok'{}, delete(Config, Server2, Q2)),
 
@@ -2192,9 +2200,11 @@ leader_locator_client_local(Config) ->
                  declare(Config, Server3, Q3, [{<<"x-queue-type">>, longstr, <<"stream">>},
                                               {<<"x-queue-leader-locator">>, longstr, <<"client-local">>}])),
 
-
-    ?assertEqual(Server3, proplists:get_value(leader,
-                                              find_queue_info(Q3, Config, 0, [leader]))),
+    await_condition(
+      fun () ->
+              Server3 == proplists:get_value(leader,
+                                             find_queue_info(Q3, Config, 0, [leader]))
+      end, 60),
 
     ?assertMatch(#'queue.delete_ok'{}, delete(Config, Server3, Q3)),
     ok.
@@ -2235,7 +2245,7 @@ leader_locator_balanced_maintenance(Config) ->
                  declare(Config, Server1, Q, [{<<"x-queue-type">>, longstr, <<"stream">>},
                                               {<<"x-queue-leader-locator">>, longstr, <<"balanced">>}])),
 
-    rabbit_ct_helpers:await_condition(
+    await_condition(
       fun() ->
               Info = find_queue_info(Config, [leader]),
               Leader = proplists:get_value(leader, Info),