implement rabbitmq-queues pick_member_with_highest_index <queue> cli command for quorum queues

Author: Ayanda-D

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -68,6 +68,8 @@
          notify_decorators/3,
          spawn_notify_decorators/3]).
 
+-export([get_member_with_highest_index/3]).
+
 -export([is_enabled/0,
          is_compatible/3,
          declare/2,
@@ -1245,7 +1247,7 @@ key_metrics_rpc(ServerId) ->
     Metrics = ra:key_metrics(ServerId),
     Metrics#{machine_version => rabbit_fifo:version()}.
 
--spec status(rabbit_types:vhost(), Name :: rabbit_misc:resource_name()) ->
+-spec status(rabbit_types:vhost(), rabbit_misc:resource_name()) ->
     [[{binary(), term()}]] | {error, term()}.
 status(Vhost, QueueName) ->
     %% Handle not found queues
@@ -1335,6 +1337,38 @@ get_sys_status(Proc) ->
 
     end.
 
+-spec get_member_with_highest_index(rabbit_types:vhost(), rabbit_misc:resource_name(), atom()) ->
+    [[{binary(), term()}]] | {error, term()}.
+get_member_with_highest_index(Vhost, QueueName, IndexName) ->
+    case ?MODULE:status(Vhost, QueueName) of
+        Status when is_list(Status) ->
+            IndexNameInternal = rabbit_data_coercion:to_atom(IndexName),
+            case index_name_to_status_key(IndexNameInternal) of
+                Key when is_binary(Key) ->
+                    {_HighestIndexValue, HighestEntry} =
+                        lists:foldl(
+                            fun(Entry, {PreviousIndexValue, _PreviousEntry} = Acc) ->
+                                case rabbit_misc:pget(Key, Entry) of
+                                    CurrentIndexValue when is_integer(CurrentIndexValue),
+                                                           CurrentIndexValue > PreviousIndexValue ->
+                                        {CurrentIndexValue, Entry};
+                                    _ ->
+                                        Acc
+                                end
+                            end, {-100, []}, Status),
+                    [HighestEntry];
+                undefined ->
+                    []
+            end;
+        {error, _} = Error ->
+            Error
+    end.
+
+index_name_to_status_key(I) when I =:= commit;   I =:= commit_index   -> <<"Commit Index">>;
+index_name_to_status_key(I) when I =:= log;      I =:= log_index      -> <<"Last Log Index">>;
+index_name_to_status_key(I) when I =:= snapshot; I =:= snapshot_index -> <<"Snapshot Index">>;
+index_name_to_status_key(_I) -> undefined.
+
 add_member(VHost, Name, Node, Membership, Timeout)
   when is_binary(VHost) andalso
        is_binary(Name) andalso

deps/rabbitmq_cli/test/queues/pick_member_with_highest_index_test.exs

@@ -0,0 +1,55 @@
+## This Source Code Form is subject to the terms of the Mozilla Public
+## License, v. 2.0. If a copy of the MPL was not distributed with this
+## file, You can obtain one at https://mozilla.org/MPL/2.0/.
+##
+## Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
+
+defmodule RabbitMQ.CLI.Queues.Commands.PickMemberWithHighestIndexCommandTest do
+  use ExUnit.Case, async: false
+  import TestHelper
+
+  @command RabbitMQ.CLI.Queues.Commands.PickMemberWithHighestIndexCommand
+
+  setup_all do
+    RabbitMQ.CLI.Core.Distribution.start()
+
+    :ok
+  end
+
+  setup context do
+    {:ok,
+     opts: %{
+       node: get_rabbit_hostname(),
+       timeout: context[:test_timeout] || 30000
+     }}
+  end
+
+  test "validate: when no arguments are provided, returns a failure" do
+    assert @command.validate([], %{}) == {:validation_failure, :not_enough_args}
+  end
+
+  test "validate: when two or more arguments are provided, returns a failure" do
+    assert @command.validate(["quorum-queue-a", "one-extra-arg"], %{}) ==
+             {:validation_failure, :too_many_args}
+
+    assert @command.validate(
+             ["quorum-queue-a", "extra-arg", "another-extra-arg"],
+             %{}
+           ) == {:validation_failure, :too_many_args}
+  end
+
+  test "validate: treats one positional arguments and default switches as a success" do
+    assert @command.validate(["quorum-queue-a"], %{}) == :ok
+  end
+
+  @tag test_timeout: 3000
+  test "run: targeting an unreachable node throws a badrpc" do
+    assert match?(
+             {:badrpc, _},
+             @command.run(
+               ["quorum-queue-a"],
+               %{node: :jake@thedog, vhost: "/", index: "log", timeout: 200}
+             )
+           )
+  end
+end