Use node with latest machine version to connect

References #4133

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -260,9 +260,7 @@ register_listener(Q) when ?is_amqqueue(Q)->
     #{name := StreamId} = amqqueue:get_type_state(Q),
     process_command({register_listener,
                      #{pid => self(),
-                       node => node(self()),
-                       stream_id => StreamId,
-                       type => leader}}).
+                       stream_id => StreamId}}).
 
 -spec register_local_member_listener(amqqueue:amqqueue()) ->
     {error, term()} | {ok, ok | stream_not_found, atom() | {atom(), atom()}}.
@@ -462,11 +460,22 @@ apply(#{machine_version := MachineVersion} = Meta,
 
 apply(#{machine_version := MachineVersion} = Meta,
       {register_listener, #{pid := Pid,
-                            node := Node,
-                            stream_id := StreamId,
-                            type := Type}},
+                            stream_id := StreamId} = Args},
       #?MODULE{streams = Streams,
                monitors = Monitors0} = State0) when MachineVersion >= 2 ->
+    Node = case Args of
+               #{node := N} ->
+                   N;
+               _ ->
+                   node(Pid)
+           end,
+    Type = case Args of
+               #{type := T} ->
+                   T;
+               _ ->
+                   leader
+           end,
+
     case Streams of
         #{StreamId := #stream{listeners = Listeners0} = Stream0} ->
             {LKey, LValue} =

deps/rabbit/test/rabbit_stream_queue_SUITE.erl

@@ -54,7 +54,8 @@ groups() ->
      {cluster_size_3_1, [], [shrink_coordinator_cluster]},
      {cluster_size_3_2, [], [recover,
                              declare_with_node_down]},
-     {cluster_size_3_parallel_1, [parallel], [delete_replica,
+     {cluster_size_3_parallel_1, [parallel], [
+                                              delete_replica,
                                               delete_last_replica,
                                               delete_classic_replica,
                                               delete_quorum_replica,
@@ -65,7 +66,8 @@ groups() ->
                                               leader_locator_client_local,
                                               declare_delete_same_stream,
                                               leader_locator_random,
-                                              leader_locator_least_leaders]},
+                                              leader_locator_least_leaders
+                                             ]},
      {cluster_size_3_parallel_2, [parallel], all_tests()},
      {unclustered_size_3_1, [], [add_replica]},
      {unclustered_size_3_2, [], [consume_without_local_replica]},
@@ -1322,7 +1324,7 @@ consume_from_relative_time_offset(Config) ->
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
 
 consume_from_replica(Config) ->
-    [Server1, Server2 | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    [Server1, _, Server3] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
 
     Ch1 = rabbit_ct_client_helpers:open_channel(Config, Server1),
     Q = ?config(queue_name, Config),
@@ -1338,7 +1340,7 @@ consume_from_replica(Config) ->
               length(proplists:get_value(online, Info)) == 3
       end),
 
-    Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server2),
+    Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server3),
     qos(Ch2, 10, false),
 
     subscribe(Ch2, Q, false, 0),