QQ: Make rabbit_fifo_client:stat/2 backwards compatible.

Author: kjnilsson

deps/rabbit/src/rabbit_fifo_client.erl

@@ -521,10 +521,16 @@ stat(Leader) ->
 stat(Leader, Timeout) ->
     %% short timeout as we don't want to spend too long if it is going to
     %% fail anyway
-    case ra:local_query(Leader, fun rabbit_fifo:query_stat/1, Timeout) of
-      {ok, {_, {R, C}}, _} -> {ok, R, C};
-      {error, _} = Error   -> Error;
-      {timeout, _} = Error -> Error
+    %% TODO: the overview is too large to be super efficient
+    %% but we use it for backwards compatibilty
+    case ra:member_overview(Leader, Timeout) of
+      {ok, #{machine := #{num_ready_messages := R,
+                          num_checked_out := C}}, _} ->
+            {ok, R, C};
+      {error, _} = Error ->
+            Error;
+      {timeout, _} = Error ->
+            Error
     end.
 
 update_machine_state(Server, Conf) ->

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1166,9 +1166,12 @@ stat(Q, Timeout) when ?is_amqqueue(Q) ->
     Leader = amqqueue:get_pid(Q),
     try
         case rabbit_fifo_client:stat(Leader, Timeout) of
-          {ok, _, _} = Success -> Success;
-          {error, _}           -> {ok, 0, 0};
-          {timeout, _}         -> {ok, 0, 0}
+          {ok, _, _} = Success ->
+                Success;
+          {error, _} ->
+                {ok, 0, 0};
+          {timeout, _} ->
+                {ok, 0, 0}
         end
     catch
         _:_ ->

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -3767,9 +3767,9 @@ queue_length_limit_policy_cleared(Config) ->
             {<<"overflow">>, <<"reject-publish">>}]),
     timer:sleep(1000),
     RaName = ra_name(QQ),
-    QueryFun = fun rabbit_fifo:overview/1,
-    ?awaitMatch({ok, {_, #{config := #{max_length := 2}}}, _},
-                rpc:call(Server, ra, local_query, [RaName, QueryFun]),
+    % QueryFun = fun rabbit_fifo:overview/1,
+    ?awaitMatch({ok, #{machine := #{config := #{max_length := 2}}}, _},
+                ra:member_overview({RaName, Server}),
                 ?DEFAULT_AWAIT),
     #'confirm.select_ok'{} = amqp_channel:call(Ch, #'confirm.select'{}),
     ok = publish_confirm(Ch, QQ),
@@ -3778,8 +3778,8 @@ queue_length_limit_policy_cleared(Config) ->
     wait_for_messages_ready(Servers, RaName, 3),
     fail = publish_confirm(Ch, QQ),
     ok = rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"max-length">>),
-    ?awaitMatch({ok, {_, #{config := #{max_length := undefined}}}, _},
-                rpc:call(Server, ra, local_query, [RaName, QueryFun]),
+    ?awaitMatch({ok, #{machine := #{config := #{max_length := undefined}}}, _},
+                ra:member_overview({RaName, Server}),
                 ?DEFAULT_AWAIT),
     ok = publish_confirm(Ch, QQ),
     wait_for_messages_ready(Servers, RaName, 4).