add --offline-members flag to control including down members in rabbitmq-queues member_with_highest_index command

Author: Ayanda-D

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -68,7 +68,7 @@
          notify_decorators/3,
          spawn_notify_decorators/3]).
 
--export([get_member_with_highest_index/3]).
+-export([get_member_with_highest_index/4]).
 
 -export([is_enabled/0,
          is_compatible/3,
@@ -1337,9 +1337,9 @@ get_sys_status(Proc) ->
 
     end.
 
--spec get_member_with_highest_index(rabbit_types:vhost(), rabbit_misc:resource_name(), atom()) ->
+-spec get_member_with_highest_index(rabbit_types:vhost(), rabbit_misc:resource_name(), atom(), boolean()) ->
     [[{binary(), term()}]] | {error, term()}.
-get_member_with_highest_index(Vhost, QueueName, IndexName) ->
+get_member_with_highest_index(Vhost, QueueName, IndexName, IncludeOfflineMembers) ->
     case ?MODULE:status(Vhost, QueueName) of
         Status when is_list(Status) ->
             IndexNameInternal = rabbit_data_coercion:to_atom(IndexName),
@@ -1348,9 +1348,14 @@ get_member_with_highest_index(Vhost, QueueName, IndexName) ->
                     {_HighestIndexValue, HighestEntry} =
                         lists:foldl(
                             fun(Entry, {PreviousIndexValue, _PreviousEntry} = Acc) ->
-                                case rabbit_misc:pget(Key, Entry) of
-                                    CurrentIndexValue when is_integer(CurrentIndexValue),
-                                                           CurrentIndexValue > PreviousIndexValue ->
+                                State = rabbit_misc:pget(<<"Raft State">>, Entry),
+                                case {rabbit_misc:pget(Key, Entry), IncludeOfflineMembers} of
+                                    {CurrentIndexValue, false} when is_integer(CurrentIndexValue),
+                                                                    CurrentIndexValue > PreviousIndexValue,
+                                                                    State /= noproc ->
+                                        {CurrentIndexValue, Entry};
+                                    {CurrentIndexValue, true} when is_integer(CurrentIndexValue),
+                                                                   CurrentIndexValue > PreviousIndexValue ->
                                         {CurrentIndexValue, Entry};
                                     _ ->
                                         Acc

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -4591,8 +4591,8 @@ get_member_with_highest_index(Config) ->
         [{Node1, leader,   1015, 1010, 1010}, %% highest SnapshotIdx
          {Node2, follower, 1015, 1010, 1010}, %% highest SnapshotIdx (duplicate)
          {Node3, follower, 1013, 1013, 1009}, %% highest CommitIdx
-         {Node4, follower, 1016, 1009, 1008}, %% highest LogIdx
-         {Node5, follower, 1013, 1012, undefined}],
+         {Node4, follower, 1016, 1009, undefined}, %% highest LogIdx
+         {Node5, noproc,   1050, 1050, 1050}], %% highest but noproc
 
     Term = 1,
     MachineVersion = 7,
@@ -4622,7 +4622,7 @@ get_member_with_highest_index(Config) ->
           {<<"Last Written">>,1016},
           {<<"Last Applied">>,1016},
           {<<"Commit Index">>, 1009},
-          {<<"Snapshot Index">>, 1008},
+          {<<"Snapshot Index">>, undefined},
           {<<"Term">>, Term},
           {<<"Machine Version">>, MachineVersion}]],
 
@@ -4667,6 +4667,21 @@ get_member_with_highest_index(Config) ->
     [?assertEqual(ExpectedHighestSnapshotIdx,
         rabbit_quorum_queue:get_member_with_highest_index(VHost, Q, I)) || I <- [snapshot, snapshot_index]],
 
+    ExpectedHighestIdxForAll =
+        [[{<<"Node Name">>, Node5},
+          {<<"Raft State">>, noproc},
+          {<<"Last Log Index">>, 1050},
+          {<<"Last Written">>,1050},
+          {<<"Last Applied">>,1050},
+          {<<"Commit Index">>, 1050},
+          {<<"Snapshot Index">>, 1050},
+          {<<"Term">>, Term},
+          {<<"Machine Version">>, MachineVersion}]],
+
+    [?assertEqual(ExpectedHighestIdxForAll,
+        rabbit_quorum_queue:get_member_with_highest_index(VHost, Q, I, true))
+            || I <- [log, log_index, commit, commit_index, snapshot, snapshot_index]],
+
     ok.
 
 leader_locator_client_local(Config) ->

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

@@ -13,18 +13,19 @@ defmodule RabbitMQ.CLI.Queues.Commands.MemberWithHighestIndexCommand do
   use RabbitMQ.CLI.Core.AcceptsOnePositionalArgument
   use RabbitMQ.CLI.Core.RequiresRabbitAppRunning
 
-  def switches(), do: [index: :string, timeout: :integer]
-  def aliases(), do: [i: :index, t: :timeout]
+  def switches(), do: [offline_members: :boolean, index: :string, timeout: :integer]
+  def aliases(), do: [o: :offline_members, i: :index, t: :timeout]
 
   def merge_defaults(args, opts) do
-    {args, Map.merge(%{vhost: "/", index: "commit"}, opts)}
+    {args, Map.merge(%{vhost: "/", index: "commit", offline_members: true}, opts)}
   end
 
-  def run([name] = _args, %{vhost: vhost, index: index, node: node_name}) do
+  def run([name] = _args, %{vhost: vhost, index: index, node: node_name, offline_members: offline_members}) do
     case :rabbit_misc.rpc_call(node_name, :rabbit_quorum_queue, :get_member_with_highest_index, [
            vhost,
            name,
-           to_atom(String.downcase(index))
+           to_atom(String.downcase(index)),
+           offline_members
          ]) do
       {:error, :classic_queue_not_supported} ->
         index = format_index(String.downcase(index))
@@ -42,11 +43,12 @@ defmodule RabbitMQ.CLI.Queues.Commands.MemberWithHighestIndexCommand do
 
   def formatter(), do: RabbitMQ.CLI.Formatters.PrettyTable
 
-  def usage, do: "member_with_highest_index <queue> [--vhost <vhost>] [--index <commit|commit_index|log|log_index|snapshot|snapshot_index>]"
+  def usage, do: "member_with_highest_index <queue> [--vhost <vhost>] [--offline-members] [--index <commit|commit_index|log|log_index|snapshot|snapshot_index>]"
 
   def usage_additional do
     [
       ["<queue>", "quorum queue name"],
+      ["--offline-members", "include members which are down (in noproc state)"],
       ["--index <commit|commit_index|log|log_index|snapshot|snapshot_index>", "name of the index to use to lookup highest member"]
     ]
   end

deps/rabbitmq_cli/test/queues/member_with_highest_index_test.exs

@@ -48,7 +48,7 @@ defmodule RabbitMQ.CLI.Queues.Commands.MemberWithHighestIndexCommandTest do
              {:badrpc, _},
              @command.run(
                ["quorum-queue-a"],
-               %{node: :jake@thedog, vhost: "/", index: "log", timeout: 200}
+               %{node: :jake@thedog, vhost: "/", index: "log", offline_members: true, timeout: 200}
              )
            )
   end