Make SAC disconnect timeout configurable

Author: acogoluegnes

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -512,7 +512,7 @@ start_coordinator_cluster() ->
                      "initial machine version ~b",
                      [Nodes, MinVersion]),
     case ra:start_cluster(?RA_SYSTEM,
-                          [make_ra_conf(Node, Nodes, MinVersion)
+                          [make_system_conf(Node, Nodes, MinVersion)
                            || Node <- Nodes]) of
         {ok, Started, _} ->
             rabbit_log:debug("Started stream coordinator on ~w", [Started]),
@@ -536,7 +536,11 @@ version() -> 5.
 which_module(_) ->
     ?MODULE.
 
-init(_Conf) ->
+init(#{machine_version := MacVersion} = Conf) when MacVersion >= 5 ->
+    SacConf = maps:get(sac_conf, Conf, #{}),
+    #?MODULE{single_active_consumer =
+             rabbit_stream_sac_coordinator:init_state(SacConf)};
+init(_) ->
     #?MODULE{single_active_consumer = rabbit_stream_sac_coordinator_v4:init_state()}.
 
 -spec apply(ra_machine:command_meta_data(), command(), state()) ->
@@ -799,7 +803,8 @@ all_member_nodes(Streams) ->
         end, #{}, Streams)).
 
 tick(_Ts, _State) ->
-    [{aux, maybe_resize_coordinator_cluster}].
+    [{aux, maybe_resize_coordinator_cluster},
+     {aux, maybe_update_sac_configuration}].
 
 members() ->
     %% TODO: this can be replaced with a ra_leaderboard
@@ -876,15 +881,27 @@ maybe_handle_stale_nodes(MemberNodes, ExpectedNodes,
             rabbit_log:debug("Stale nodes detected in stream SAC "
                              "coordinator: ~w. Purging state.",
                              [Stale]),
-            Mod = sac_module(MachineVersion),
-            ra:pipeline_command(LeaderPid, Mod:make_purge_nodes(Stale)),
+            ra:pipeline_command(LeaderPid, sac_make_purge_nodes(Stale)),
             ok;
         _ ->
             ok
     end;
 maybe_handle_stale_nodes(_, _, _, _) ->
     ok.
 
+maybe_update_sac_configuration(RaAux, MacVersion) when MacVersion >= 5 ->
+    #?MODULE{single_active_consumer = SacState} = ra_aux:machine_state(RaAux),
+    case sac_check_conf_change(SacState) of
+        {new, UpdatedConf} ->
+            Leader = ra_aux:leader_id(RaAux),
+            ra:pipeline_command(Leader, sac_make_update_conf(UpdatedConf)),
+            ok;
+        _ ->
+            ok
+    end;
+maybe_update_sac_configuration(_, _) ->
+    ok.
+
 add_member(Members, Node) ->
     MinMacVersion = erpc:call(Node, ?MODULE, version, []),
     Conf = make_ra_conf(Node, [N || {_, N} <- Members], MinMacVersion),
@@ -967,6 +984,11 @@ handle_aux(leader, _, maybe_resize_coordinator_cluster,
            AuxState, RaAux) ->
     %% Coordinator resizing is still happening, let's ignore this tick event
     {no_reply, AuxState, RaAux};
+handle_aux(leader, _, maybe_update_sac_configuration,
+           AuxState, RaAux) ->
+    MachineVersion = ra_aux:effective_machine_version(RaAux),
+    maybe_update_sac_configuration(RaAux, MachineVersion),
+    {no_reply, AuxState, RaAux};
 handle_aux(leader, _, {down, Pid, _},
            #aux{resizer = Pid} = Aux, RaAux) ->
     %% Coordinator resizing has finished
@@ -1328,6 +1350,11 @@ phase_update_mnesia(StreamId, Args, #{reference := QName,
 format_ra_event(ServerId, Evt) ->
     {stream_coordinator_event, ServerId, Evt}.
 
+make_system_conf(Node, Nodes, MinMacVersion) ->
+    RaConf = make_ra_conf(Node, Nodes, MinMacVersion),
+    SacConf = make_sac_conf(MinMacVersion),
+    RaConf#{sac_conf => SacConf}.
+
 make_ra_conf(Node, Nodes, MinMacVersion) ->
     UId = ra:new_uid(ra_lib:to_binary(?MODULE)),
     Formatter = {?MODULE, format_ra_event, []},
@@ -1346,6 +1373,11 @@ make_ra_conf(Node, Nodes, MinMacVersion) ->
       initial_machine_version => MinMacVersion,
       ra_event_formatter => Formatter}.
 
+make_sac_conf(MinMacVersion) when MinMacVersion >= 5 ->
+    sac_make_conf();
+make_sac_conf(_) ->
+    #{}.
+
 filter_command(_Meta, {delete_replica, _, #{node := Node}}, #stream{id = StreamId,
                                                                     members = Members0}) ->
     Members = maps:filter(fun(_, #member{target = S}) when S =/= deleted ->
@@ -2420,3 +2452,15 @@ maps_to_list(M) ->
 
 ra_local_query(QueryFun) ->
     ra:local_query({?MODULE, node()}, QueryFun, infinity).
+
+sac_make_conf() ->
+    rabbit_stream_sac_coordinator:make_conf().
+
+sac_make_purge_nodes(Nodes) ->
+    rabbit_stream_sac_coordinator:make_purge_nodes(Nodes).
+
+sac_make_update_conf(Conf) ->
+    rabbit_stream_sac_coordinator:make_update_conf(Conf).
+
+sac_check_conf_change(SacState) ->
+    rabbit_stream_sac_coordinator:check_conf_change(SacState).

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -22,7 +22,8 @@
                      #command_unregister_consumer{} |
                      #command_activate_consumer{} |
                      #command_connection_reconnected{} |
-                     #command_purge_nodes{}.
+                     #command_purge_nodes{} |
+                     #command_update_conf{}.
 
 -opaque state() :: #?MODULE{}.
 
@@ -37,7 +38,7 @@
          group_consumers/4,
          connection_reconnected/1]).
 -export([apply/2,
-         init_state/0,
+         init_state/1,
          send_message/2,
          ensure_monitors/4,
          handle_connection_down/2,
@@ -47,8 +48,11 @@
          consumer_groups/3,
          group_consumers/5,
          overview/1,
-         import_state/2]).
--export([make_purge_nodes/1]).
+         import_state/2,
+         make_conf/0,
+         check_conf_change/1]).
+-export([make_purge_nodes/1,
+         make_update_conf/1]).
 
 %% exported for unit tests only
 -ifdef(TEST).
@@ -70,6 +74,8 @@
 -define(DISCONN_ACT, {?DISCONNECTED, ?ACTIVE}).
 -define(FORG_ACT, {?FORGOTTTEN, ?ACTIVE}).
 
+-define(DISCONNECTED_TIMEOUT_APP_KEY, stream_sac_disconnected_timeout).
+-define(DISCONNECTED_TIMEOUT_CONF_KEY, disconnected_timeout).
 -define(DISCONNECTED_TIMEOUT_MS, 60_000).
 
 %% Single Active Consumer API
@@ -195,9 +201,12 @@ overview(#?MODULE{groups = Groups}) ->
                  Groups),
     #{num_groups => map_size(Groups), groups => GroupsOverview}.
 
--spec init_state() -> state().
-init_state() ->
-    #?MODULE{groups = #{}, pids_groups = #{}}.
+-spec init_state(map()) -> state().
+init_state(Conf) ->
+    DisconTimeout = maps:get(?DISCONNECTED_TIMEOUT_CONF_KEY, Conf,
+                             ?DISCONNECTED_TIMEOUT_MS),
+    #?MODULE{groups = #{}, pids_groups = #{},
+             conf = #{?DISCONNECTED_TIMEOUT_CONF_KEY => DisconTimeout}}.
 
 -spec apply(command(), state()) ->
                {state(), term(), ra_machine:effects()}.
@@ -319,7 +328,9 @@ apply(#command_purge_nodes{nodes = Nodes}, State0) ->
                                         {S1, Eff1} = purge_node(N, S0),
                                         {S1, Eff1 ++ Eff0}
                                 end, {State0, []}, Nodes),
-    {State1, ok, Eff}.
+    {State1, ok, Eff};
+apply(#command_update_conf{conf = NewConf}, State) ->
+    {State#?MODULE{conf = NewConf}, ok, []}.
 
 purge_node(Node, #?MODULE{groups = Groups0} = State0) ->
     PidsGroups = compute_node_pid_group_dependencies(Node, Groups0),
@@ -725,8 +736,7 @@ handle_connection_node_disconnected(ConnPid,
                                   handle_group_after_connection_node_disconnected(
                                     ConnPid, Acc, G)
                           end, State1, Groups),
-            T = application:get_env(rabbit, stream_sac_disconnected_timeout,
-                                    ?DISCONNECTED_TIMEOUT_MS),
+            T = disconnected_timeout(State2),
             {State2, [{timer, {sac, node_disconnected,
                                #{connection_pid => ConnPid}}, T}]}
     end.
@@ -850,12 +860,43 @@ handle_group_after_connection_node_disconnected(ConnPid,
 -spec import_state(ra_machine:version(), map()) -> state().
 import_state(4, #{<<"groups">> := Groups, <<"pids_groups">> := PidsGroups}) ->
     #?MODULE{groups = map_to_groups(Groups),
-             pids_groups = map_to_pids_groups(PidsGroups)}.
+             pids_groups = map_to_pids_groups(PidsGroups),
+             conf = #{disconnected_timeout => ?DISCONNECTED_TIMEOUT_MS}}.
+
+-spec make_conf() -> conf().
+make_conf() ->
+    #{?DISCONNECTED_TIMEOUT_CONF_KEY => lookup_disconnected_timeout()}.
+
+-spec check_conf_change(state()) -> {new, conf()} | unchanged.
+check_conf_change(#?MODULE{conf = Conf}) ->
+    DisconTimeout = lookup_disconnected_timeout(),
+    case Conf of
+        #{?DISCONNECTED_TIMEOUT_CONF_KEY := DT}
+          when DT /= DisconTimeout ->
+            {new, #{?DISCONNECTED_TIMEOUT_CONF_KEY => DisconTimeout}};
+        C when is_map_key(?DISCONNECTED_TIMEOUT_CONF_KEY, C) == false ->
+            {new, #{?DISCONNECTED_TIMEOUT_CONF_KEY => DisconTimeout}};
+        _ ->
+            unchanged
+    end.
 
 - spec make_purge_nodes([node()]) -> {sac, command()}.
 make_purge_nodes(Nodes) ->
     wrap_cmd(#command_purge_nodes{nodes = Nodes}).
 
+- spec make_update_conf(conf()) -> {sac, command()}.
+make_update_conf(Conf) ->
+    wrap_cmd(#command_update_conf{conf = Conf}).
+
+lookup_disconnected_timeout() ->
+    application:get_env(rabbit, ?DISCONNECTED_TIMEOUT_APP_KEY,
+                        ?DISCONNECTED_TIMEOUT_MS).
+
+disconnected_timeout(#?MODULE{conf = #{?DISCONNECTED_TIMEOUT_CONF_KEY := T}}) ->
+    T;
+disconnected_timeout(_) ->
+    ?DISCONNECTED_TIMEOUT_MS.
+
 map_to_groups(Groups) when is_map(Groups) ->
     maps:fold(fun(K, V, Acc) ->
                       Acc#{K => map_to_group(V)}

deps/rabbit/src/rabbit_stream_sac_coordinator.hrl

@@ -24,6 +24,7 @@
 -type owner() :: binary().
 -type consumer_status() :: active | waiting | deactivating.
 -type consumer_connectivity() :: connected | disconnected | forgotten.
+-type conf() :: map().
 
 -record(consumer,
         {pid :: pid(),
@@ -35,6 +36,7 @@
 -record(rabbit_stream_sac_coordinator,
         {groups :: groups(),
          pids_groups :: pids_groups(),
+         conf :: conf(),
          %% future extensibility
          reserved_1,
          reserved_2}).
@@ -66,3 +68,5 @@
         {pid :: connection_pid()}).
 -record(command_purge_nodes,
         {nodes :: [node()]}).
+-record(command_update_conf,
+        {conf :: conf()}).

deps/rabbit/test/rabbit_stream_sac_coordinator_SUITE.erl

@@ -61,6 +61,17 @@ end_per_testcase(_TestCase, _Config) ->
     meck:unload(),
     ok.
 
+check_conf_test(_) ->
+    K = disconnected_timeout,
+    Def = 60_000,
+    ?assertMatch({new, #{K := Def}},
+                 ?MOD:check_conf_change(state_with_conf(#{}))),
+    ?assertMatch({new, #{K := Def}},
+                 ?MOD:check_conf_change(state_with_conf(#{K => 42}))),
+    ?assertMatch(unchanged,
+                 ?MOD:check_conf_change(state_with_conf(#{K => Def}))),
+    ok.
+
 simple_sac_test(_) ->
     Stream = <<"stream">>,
     ConsumerName = <<"app">>,
@@ -1503,6 +1514,9 @@ state(Groups) ->
 state(Groups, PidsGroups) ->
     #?STATE{groups = Groups, pids_groups = PidsGroups}.
 
+state_with_conf(Conf) ->
+    #?STATE{conf = Conf}.
+
 register_consumer_command(Stream,
                           PartitionIndex,
                           ConsumerName,

deps/rabbitmq_stream/test/rabbit_stream_partitions_SUITE.erl

@@ -63,13 +63,13 @@ init_per_group(Group, Config) ->
                                                {aten,
                                                 [{poll_interval,
                                                   1000}]})
+       end,
+       fun(StepConfig) ->
+               rabbit_ct_helpers:merge_app_env(StepConfig,
+                                               {rabbit,
+                                                [{stream_sac_disconnected_timeout,
+                                                  2000}]})
        end]
-       % fun(StepConfig) ->
-       %         rabbit_ct_helpers:merge_app_env(StepConfig,
-       %                                         {rabbit,
-       %                                          [{stream_sac_disconnected_timeout,
-       %                                            2000}]})
-       % end]
       ++ rabbit_ct_broker_helpers:setup_steps()).
 
 end_per_group(_, Config) ->
@@ -110,6 +110,7 @@ simple_sac_consumer_should_get_disconnected_on_partition(Config) ->
     rabbit_ct_broker_helpers:block_traffic_between(F1, F2),
 
     wait_for_disconnected_consumer(Config, S),
+    wait_for_forgotten_consumer(Config, S),
 
     rabbit_ct_broker_helpers:allow_traffic_between(F1, L),
     rabbit_ct_broker_helpers:allow_traffic_between(F1, F2),