Fix stream coordinator v1-to-v2 machine state change

Monitors were not converted.

References #4133

(cherry picked from commit df61e44)
(cherry picked from commit 1ec796b)

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -73,6 +73,7 @@
                    {delete_replica, stream_id(), #{node := node()}} |
                    {policy_changed, stream_id(), #{queue := amqqueue:amqqueue()}} |
                    {register_listener, #{pid := pid(),
+                                         node := node(),
                                          stream_id := stream_id(),
                                          type := leader | local_member}} |
                    {action_failed, stream_id(), #{index := ra:index(),
@@ -524,13 +525,27 @@ apply(Meta, {nodeup, Node} = Cmd,
                   end, {Streams0, Effects0}, Streams0),
     return(Meta, State#?MODULE{monitors = Monitors,
                                streams = Streams}, ok, Effects);
-apply(Meta, {machine_version, 1, 2}, State = #?MODULE{streams = Streams0}) ->
-    Streams1 = maps:fold(fun(ListPid, LeaderPid, Acc) ->
-                            Acc#{{ListPid, leader} => LeaderPid}
+apply(Meta, {machine_version, From = 1, To = 2}, State = #?MODULE{streams = Streams0,
+                                                                  monitors = Monitors0}) ->
+    rabbit_log:info("Stream coordinator machine version changes from ~p to ~p, updating state.",
+                    [From, To]),
+    Streams1 = maps:fold(fun(S, #stream{listeners = L0} = S0, StreamAcc) ->
+                                 L1 = maps:fold(fun(ListPid, LeaderPid, LAcc) ->
+                                                        LAcc#{{ListPid, leader} => LeaderPid}
+                                                end, #{}, L0),
+                                 StreamAcc#{S => S0#stream{listeners = L1}}
                          end, #{}, Streams0),
+    Monitors1 = maps:fold(fun(P, {StreamId, listener}, Acc) ->
+                                  Acc#{P => {#{StreamId => ok}, listener}};
+                             (P, V, Acc) ->
+                                  Acc#{P => V}
+                          end, #{}, Monitors0),
     return(Meta, State#?MODULE{streams = Streams1,
+                               monitors = Monitors1,
                                listeners = undefined}, ok, []);
-apply(Meta, {machine_version, _From, _To}, State) ->
+apply(Meta, {machine_version, From, To}, State) ->
+    rabbit_log:info("Stream coordinator machine version changes from ~p to ~p, no state changes required.",
+                    [From, To]),
     return(Meta, State, ok, []);
 apply(Meta, UnkCmd, State) ->
     rabbit_log:debug("~s: unknown command ~W",

deps/rabbit/src/rabbit_stream_coordinator.hrl

@@ -57,7 +57,9 @@
                 }).
 
 -record(rabbit_stream_coordinator, {streams = #{} :: #{stream_id() => #stream{}},
-                                    monitors = #{} :: #{pid() => {stream_id(), monitor_role()}},
+                                    monitors = #{} :: #{pid() => {stream_id() | %% v0 & v1
+                                                                  #{stream_id() => ok}, %% v2
+                                                                  monitor_role()}},
                                     %% not used as of v2
                                     listeners = #{} :: undefined | #{stream_id() =>
                                                                      #{pid() := queue_ref()}},

deps/rabbit/test/rabbit_stream_coordinator_SUITE.erl

@@ -25,6 +25,7 @@ all() ->
 all_tests() ->
     [
      listeners,
+     machine_version_from_1_to_2,
      new_stream,
      leader_down,
      leader_down_scenario_1,
@@ -195,6 +196,27 @@ listeners(_) ->
 
     ok.
 
+machine_version_from_1_to_2(_) ->
+    S = <<"stream">>,
+    LeaderPid = spawn(fun() -> ok end),
+    ListPid = spawn(fun() -> ok end),
+    State0 = #?STATE{streams = #{S =>
+                                 #stream{listeners = #{ListPid => LeaderPid}}},
+                     monitors = #{ListPid => {S, listener}}},
+
+    {State1, ok, []} = apply_cmd(#{index => 42}, {machine_version, 1, 2}, State0),
+
+    Stream1 = maps:get(S, State1#?STATE.streams),
+    ?assertEqual(
+       #{{ListPid, leader} => LeaderPid},
+       Stream1#stream.listeners
+      ),
+    ?assertEqual(
+       #{ListPid => {#{S => ok}, listener}},
+       State1#?STATE.monitors
+      ),
+    ok.
+
 new_stream(_) ->
     [N1, N2, N3] = Nodes = [r@n1, r@n2, r@n3],
     StreamId = atom_to_list(?FUNCTION_NAME),