Fix force_checkpoint tests and CLI command

Author: aaron-seo

deps/rabbit/src/rabbit_fifo.erl

@@ -925,26 +925,6 @@ which_module(3) -> rabbit_fifo_v3;
 which_module(4) -> ?MODULE;
 which_module(5) -> ?MODULE.
 
--define(AUX, aux_v3).
-
--record(checkpoint, {index :: ra:index(),
-                     timestamp :: milliseconds(),
-                     smallest_index :: undefined | ra:index(),
-                     messages_total :: non_neg_integer(),
-                     indexes = ?CHECK_MIN_INDEXES :: non_neg_integer(),
-                     unused_1 = ?NIL}).
--record(aux_gc, {last_raft_idx = 0 :: ra:index()}).
--record(aux, {name :: atom(),
-              capacity :: term(),
-              gc = #aux_gc{} :: #aux_gc{}}).
--record(?AUX, {name :: atom(),
-               last_decorators_state :: term(),
-               capacity :: term(),
-               gc = #aux_gc{} :: #aux_gc{},
-               tick_pid :: undefined | pid(),
-               cache = #{} :: map(),
-               last_checkpoint :: #checkpoint{}}).
-
 init_aux(Name) when is_atom(Name) ->
     %% TODO: catch specific exception throw if table already exists
     ok = ra_machine_ets:create_table(rabbit_fifo_usage,

deps/rabbit/src/rabbit_fifo.hrl

@@ -227,3 +227,23 @@
                     msg_ttl => non_neg_integer(),
                     created => non_neg_integer()
                    }.
+
+-define(AUX, aux_v3).
+
+-record(checkpoint, {index :: ra:index(),
+                     timestamp :: milliseconds(),
+                     smallest_index :: undefined | ra:index(),
+                     messages_total :: non_neg_integer(),
+                     indexes = ?CHECK_MIN_INDEXES :: non_neg_integer(),
+                     unused_1 = ?NIL}).
+-record(aux_gc, {last_raft_idx = 0 :: ra:index()}).
+-record(aux, {name :: atom(),
+              capacity :: term(),
+              gc = #aux_gc{} :: #aux_gc{}}).
+-record(?AUX, {name :: atom(),
+               last_decorators_state :: term(),
+               capacity :: term(),
+               gc = #aux_gc{} :: #aux_gc{},
+               tick_pid :: undefined | pid(),
+               cache = #{} :: map(),
+               last_checkpoint :: #checkpoint{}}).

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -2089,13 +2089,13 @@ force_all_queues_shrink_member_to_current_member(ListQQFun) when is_function(Lis
 force_checkpoint_on_queue(QName) ->
     Node = node(),
     QNameFmt = rabbit_misc:rs(QName),
-    case rabbit_amqqueue:lookup(QName) of
+    case rabbit_db_queue:get_durable(QName) of
         {ok, Q} when ?amqqueue_is_classic(Q) ->
             {error, classic_queue_not_supported};
         {ok, Q} when ?amqqueue_is_quorum(Q) ->
             {RaName, _} = amqqueue:get_pid(Q),
-            rpc:call(Node, ra, cast_aux_command, [{RaName, Node}, force_checkpoint]),
-            rabbit_log:debug("Sent command to force checkpoint ~ts", [QNameFmt]);
+            rabbit_log:debug("Sending command to force ~ts to take a checkpoint", [QNameFmt]),
+            rpc:call(Node, ra, cast_aux_command, [{RaName, Node}, force_checkpoint]);
         {ok, _Q} ->
             {error, not_quorum_queue};
         {error, _} = E ->
@@ -2114,8 +2114,7 @@ force_checkpoint(VhostSpec, QueueSpec) ->
                  {QName, {error, Err}}
          end
      end
-     || Q <- rabbit_amqqueue:list(),
-        amqqueue:get_type(Q) == ?MODULE,
+     || Q <- rabbit_db_queue:get_all_durable_by_type(?MODULE),
         is_match(amqqueue:get_vhost(Q), VhostSpec)
         andalso is_match(get_resource_name(amqqueue:get_name(Q)), QueueSpec)].
 
@@ -2179,4 +2178,3 @@ file_handle_other_reservation() ->
 
 file_handle_release_reservation() ->
     ok.
-

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -10,6 +10,7 @@
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
 -include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
+-include_lib("rabbit/src/rabbit_fifo.hrl").
 
 -import(queue_utils, [wait_for_messages_ready/3,
                       wait_for_messages_pending_ack/3,
@@ -1327,12 +1328,9 @@ force_checkpoint_on_queue(Config) ->
     % Wait for initial checkpoint and make sure it's 0; checkpoint hasn't been triggered yet.
     rabbit_ct_helpers:await_condition(
       fun() ->
-          {ok, #{aux := Aux1}, _} = rpc:call(Server0, ra, member_overview, [{RaName, Server0}]),
-          {aux_v3, _, _, _, _, _, _, {checkpoint, Index, _, _, _, _, _}} = Aux1,
-          case Index of
-              0 -> true;
-              _ -> false
-          end
+          {ok, #{aux := Aux}, _} = rpc:call(Server0, ra, member_overview, [{RaName, Server0}]),
+          #aux_v3{last_checkpoint = #checkpoint{index = Index}} = Aux,
+          Index =:= 0
       end),
 
     rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
@@ -1341,19 +1339,17 @@ force_checkpoint_on_queue(Config) ->
     % Wait for initial checkpoint and make sure it's not 0
     rabbit_ct_helpers:await_condition(
       fun() ->
-          {ok, #{aux := Aux1}, _} = rpc:call(Server0, ra, member_overview, [{RaName, Server0}]),
-          {aux_v3, _, _, _, _, _, _, {checkpoint, Index, _, _, _, _, _}} = Aux1,
-          case Index of
-              0 -> false;
-              _ -> true
-          end
+          {ok, #{aux := Aux}, _} = rpc:call(Server0, ra, member_overview, [{RaName, Server0}]),
+          #aux_v3{last_checkpoint = #checkpoint{index = Index}} = Aux,
+          Index =/= 0
       end).
 
 force_checkpoint(Config) ->
     [Server0, _Server1, _Server2] =
         rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server0),
     QQ = ?config(queue_name, Config),
+    QQName = rabbit_misc:r(<<"/">>, queue, QQ),
     CQ = <<"force_checkpoint_cq">>,
     RaName = ra_name(QQ),
 
@@ -1366,24 +1362,12 @@ force_checkpoint(Config) ->
     rabbit_ct_client_helpers:publish(Ch, QQ, 3),
     wait_for_messages_ready([Server0], RaName, 3),
 
-    meck:expect(rabbit_quorum_queue, force_checkpoint_on_queue, fun(Q) -> ok end),
-
-    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+    ForceCheckpointRes = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
         force_checkpoint, [<<".*">>, <<".*">>]),
-
-    % Waiting here to make sure checkpoint has been forced
-    rabbit_ct_helpers:await_condition(
-      fun() ->
-          {ok, #{aux := Aux1}, _} = rpc:call(Server0, ra, member_overview, [{RaName, Server0}]),
-          {aux_v3, _, _, _, _, _, _, {checkpoint, Index, _, _, _, _, _}} = Aux1,
-          case Index of
-              0 -> false;
-              _ -> true
-          end
-      end),
-
-    % Make sure force_checkpoint_on_queue was only called for the quorun queue
-    ?assertEqual(1, meck:num_calls(rabbit_quorum_queue, force_checkpoint_on_queue, '_')).
+    ExpectedRes = [{QQName, {ok}}],
+    
+    % Result should only have quorum queue
+    ?assertEqual(ExpectedRes, ForceCheckpointRes).
 
 % Tests that, if the process of a QQ is dead in the moment of declaring a policy
 % that affects such queue, when the process is made available again, the policy

deps/rabbitmq_cli/lib/rabbitmq/cli/queues/commands/force_checkpoint_command.ex

@@ -35,9 +35,6 @@ defmodule RabbitMQ.CLI.Queues.Commands.ForceCheckpointCommand do
     args = [vhost_pat, queue_pat]
 
     case :rabbit_misc.rpc_call(node_name, :rabbit_quorum_queue, :force_checkpoint, args) do
-      {:error, _} = error ->
-        error
-
       {:badrpc, _} = error ->
         error
 
@@ -46,15 +43,15 @@ defmodule RabbitMQ.CLI.Queues.Commands.ForceCheckpointCommand do
             do: [
               {:vhost, vhost},
               {:name, name},
-              {:result, format_result(res)}
+              {:result, res}
             ]
 
       results ->
         for {{:resource, vhost, _kind, name}, res} <- results,
             do: [
               {:vhost, vhost},
               {:name, name},
-              {:result, format_result(res)}
+              {:result, res}
             ]
     end
   end
@@ -88,20 +85,4 @@ defmodule RabbitMQ.CLI.Queues.Commands.ForceCheckpointCommand do
   def banner([], _) do
     "Forcing checkpoint for all matching quorum queues..."
   end
-
-  #
-  # Implementation
-  #
-
-  defp format_result({:ok}) do
-    "ok"
-  end
-
-  defp format_result({:error, :timeout}) do
-    "error: the operation timed out and may not have been completed"
-  end
-
-  defp format_result({:error, err}) do
-    to_string(:io_lib.format("error: ~W", [err, 10]))
-  end
 end