Merge pull request #8203 from terry-xiaoyu/force_sync_configs

Change the default retry_interval of cluster_rpc call to 1min

apps/emqx/src/emqx_authentication.erl

@@ -508,10 +508,10 @@ handle_info(Info, State) ->
 
 terminate(Reason, _State) ->
     case Reason of
-        normal ->
-            ok;
         {shutdown, _} ->
             ok;
+        Reason when Reason == normal; Reason == shutdown ->
+            ok;
         Other ->
             ?SLOG(error, #{
                 msg => "emqx_authentication_terminating",

apps/emqx_bridge/src/emqx_bridge.erl

@@ -55,15 +55,7 @@ load() ->
         fun({Type, NamedConf}) ->
             lists:foreach(
                 fun({Name, Conf}) ->
-                    _Res = emqx_bridge_resource:create(Type, Name, Conf, Opts),
+                    safe_load_bridge(Type, Name, Conf, Opts)
-                    ?tp(
-                        emqx_bridge_loaded,
-                        #{
-                            type => Type,
-                            name => Name,
-                            res => _Res
-                        }
-                    )
                 end,
                 maps:to_list(NamedConf)
             )
@@ -71,6 +63,29 @@ load() ->
         maps:to_list(Bridges)
     ).
 
+safe_load_bridge(Type, Name, Conf, Opts) ->
+    try
+        _Res = emqx_bridge_resource:create(Type, Name, Conf, Opts),
+        ?tp(
+            emqx_bridge_loaded,
+            #{
+                type => Type,
+                name => Name,
+                res => _Res
+            }
+        )
+    catch
+        Err:Reason:ST ->
+            ?SLOG(error, #{
+                msg => "load_bridge_failed",
+                type => Type,
+                name => Name,
+                error => Err,
+                reason => Reason,
+                stacktrace => ST
+            })
+    end.
+
 load_hook() ->
     Bridges = emqx:get_config([bridges], #{}),
     load_hook(Bridges).

apps/emqx_bridge/src/emqx_bridge_api.erl

@@ -571,6 +571,9 @@ pick_bridges_by_id(Type, Name, BridgesAllNodes) ->
                 [] ->
                     ?SLOG(warning, #{
                         msg => "bridge_inconsistent_in_cluster",
+                        reason => not_found,
+                        type => Type,
+                        name => Name,
                         bridge => emqx_bridge_resource:bridge_id(Type, Name)
                     }),
                     Acc

apps/emqx_conf/src/emqx_cluster_rpc.erl

@@ -547,7 +547,7 @@ commit_status_trans(Operator, TnxId) ->
     mnesia:select(?CLUSTER_COMMIT, [{MatchHead, [Guard], [Result]}]).
 
 get_retry_ms() ->
-    emqx_conf:get(["node", "cluster_call", "retry_interval"], 1000).
+    emqx_conf:get([node, cluster_call, retry_interval], timer:minutes(1)).
 
 maybe_init_tnx_id(_Node, TnxId) when TnxId < 0 -> ok;
 maybe_init_tnx_id(Node, TnxId) ->

apps/emqx_conf/src/emqx_conf_schema.erl

@@ -585,7 +585,7 @@ fields("cluster_call") ->
                 emqx_schema:duration(),
                 #{
                     desc => ?DESC(cluster_call_retry_interval),
-                    default => "1s"
+                    default => "1m"
                 }
             )},
         {"max_history",

apps/emqx_conf/test/emqx_cluster_rpc_SUITE.erl

@@ -44,6 +44,7 @@ init_per_suite(Config) ->
     application:load(emqx_conf),
     ok = ekka:start(),
     ok = mria_rlog:wait_for_shards([?CLUSTER_RPC_SHARD], infinity),
+    ok = emqx_config:put([node, cluster_call, retry_interval], 1000),
     meck:new(emqx_alarm, [non_strict, passthrough, no_link]),
     meck:expect(emqx_alarm, activate, 3, ok),
     meck:expect(emqx_alarm, deactivate, 3, ok),