Adapt to new {error, timeout} return value in Khepri 0.14.0

See rabbitmq/khepri#256.

(cherry picked from commit 88c1ad2)

Author: the-mikedavis

deps/rabbit/src/rabbit_khepri.erl

@@ -287,7 +287,9 @@ wait_for_leader(Timeout, Retries) ->
         Exists when is_boolean(Exists) ->
             rabbit_log:info("Khepri leader elected"),
             ok;
-        {error, {timeout, _ServerId}} ->
+        {error, timeout} -> %% Khepri >= 0.14.0
+            wait_for_leader(Timeout, Retries -1);
+        {error, {timeout, _ServerId}} -> %% Khepri < 0.14.0
             wait_for_leader(Timeout, Retries -1);
         {error, Reason} ->
             throw(Reason)
@@ -491,13 +493,13 @@ remove_down_member(NodeToRemove) ->
                [NodeToRemove, ?RA_CLUSTER_NAME, Reason],
                #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
             Error;
-        {timeout, _} = Reason ->
+        {timeout, _LeaderId} ->
             ?LOG_ERROR(
                "Failed to remove remote down node ~s from Khepri "
-               "cluster \"~s\": ~p",
-               [NodeToRemove, ?RA_CLUSTER_NAME, Reason],
+               "cluster \"~s\" due to timeout",
+               [NodeToRemove, ?RA_CLUSTER_NAME],
                #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-            {error, Reason}
+            {error, timeout}
     end.
 
 %% @private

deps/rabbit/test/cluster_minority_SUITE.erl

@@ -181,27 +181,27 @@ consume_from_queue(Config) ->
                  amqp_channel:call(Ch, #'basic.consume'{queue = <<"test-queue">>})).
 
 add_vhost(Config) ->
-    ?assertMatch({error, {timeout, _}},
+    ?assertMatch({error, timeout},
                  rabbit_ct_broker_helpers:add_vhost(Config, <<"vhost1">>)).
 
 update_vhost(Config) ->
-    ?assertThrow({error, {timeout, _}},
+    ?assertThrow({error, timeout},
                  rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_vhost, update_tags,
                                               [<<"/">>, [carrots], <<"user">>])).
 
 delete_vhost(Config) ->
     ?assertMatch({'EXIT', _}, rabbit_ct_broker_helpers:delete_vhost(Config, <<"vhost1">>)).
 
 add_user(Config) ->
-    ?assertMatch({error, {timeout, _}},
+    ?assertMatch({error, timeout},
                  rabbit_ct_broker_helpers:add_user(Config, <<"user1">>)).
 
 update_user(Config) ->
-    ?assertMatch({error, {timeout, _}},
+    ?assertMatch({error, timeout},
                  rabbit_ct_broker_helpers:set_user_tags(Config, 0, <<"user1">>, [<<"admin">>])).
 
 delete_user(Config) ->
-    ?assertMatch({error, {timeout, _}},
+    ?assertMatch({error, timeout},
                  rabbit_ct_broker_helpers:delete_user(Config, <<"user1">>)).
 
 set_policy(Config) ->

deps/rabbitmq_cli/lib/rabbitmq/cli/default_output.ex

@@ -68,7 +68,13 @@ defmodule RabbitMQ.CLI.DefaultOutput do
   defp normalize_output({unknown, _} = input, _opts) when is_atom(unknown), do: {:error, input}
   defp normalize_output(result, _opts) when not is_atom(result), do: {:ok, result}
 
+  defp format_khepri_output({:error, :timeout}, %{node: node_name}) do
+    # Khepri >= 0.14.0
+    {:error, RabbitMQ.CLI.Core.ExitCodes.exit_tempfail(), khepri_timeout_error(node_name)}
+  end
+
   defp format_khepri_output({:error, {:timeout, {:rabbitmq_metadata, _}}}, %{node: node_name}) do
+    # Khepri < 0.14.0
     {:error, RabbitMQ.CLI.Core.ExitCodes.exit_tempfail(), khepri_timeout_error(node_name)}
   end