Merge pull request #9889 from keynslug/fix/EMQX-8857/force-clean-start-mqtt-bridge

fix(mqtt-bridge): stop respecting `clean_start` config parameter

apps/emqx_bridge/test/emqx_bridge_mqtt_SUITE.erl

@@ -242,6 +242,29 @@ t_mqtt_conn_bridge_ingress(_) ->
 
     ok.
 
+t_mqtt_conn_bridge_ignores_clean_start(_) ->
+    BridgeName = atom_to_binary(?FUNCTION_NAME),
+    BridgeID = create_bridge(
+        ?SERVER_CONF(<<"user1">>)#{
+            <<"type">> => ?TYPE_MQTT,
+            <<"name">> => BridgeName,
+            <<"ingress">> => ?INGRESS_CONF,
+            <<"clean_start">> => false
+        }
+    ),
+
+    {ok, 200, BridgeJSON} = request(get, uri(["bridges", BridgeID]), []),
+    Bridge = jsx:decode(BridgeJSON),
+
+    %% verify that there's no `clean_start` in response
+    ?assertEqual(#{}, maps:with([<<"clean_start">>], Bridge)),
+
+    %% delete the bridge
+    {ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
+    {ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
+
+    ok.
+
 t_mqtt_conn_bridge_ingress_no_payload_template(_) ->
     User1 = <<"user1">>,
     BridgeIDIngress = create_bridge(

apps/emqx_connector/src/emqx_connector_mqtt.erl

@@ -251,7 +251,6 @@ basic_config(
         server := Server,
         proto_ver := ProtoVer,
         bridge_mode := BridgeMode,
-        clean_start := CleanStart,
         keepalive := KeepAlive,
         retry_interval := RetryIntv,
         max_inflight := MaxInflight,
@@ -271,7 +270,10 @@ basic_config(
         %% non-standard mqtt connection packets will be filtered out by LB.
         %% So let's disable bridge_mode.
         bridge_mode => BridgeMode,
-        clean_start => CleanStart,
+        %% NOTE
+        %% We are ignoring the user configuration here because there's currently no reliable way
+        %% to ensure proper session recovery according to the MQTT spec.
+        clean_start => true,
         keepalive => ms_to_s(KeepAlive),
         retry_interval => RetryIntv,
         max_inflight => MaxInflight,

apps/emqx_connector/src/mqtt/emqx_connector_mqtt_schema.erl

@@ -110,7 +110,9 @@ fields("server_configs") ->
                 boolean(),
                 #{
                     default => true,
-                    desc => ?DESC("clean_start")
+                    desc => ?DESC("clean_start"),
+                    hidden => true,
+                    deprecated => {since, "v5.0.16"}
                 }
             )},
         {keepalive, mk_duration("MQTT Keepalive.", #{default => "300s"})},

apps/emqx_connector/src/mqtt/emqx_connector_mqtt_worker.erl

@@ -425,18 +425,3 @@ printable_maps(Headers) ->
         #{},
         Headers
     ).
-
-%% TODO
-% maybe_destroy_session(#{connect_opts := ConnectOpts = #{clean_start := false}} = State) ->
-%     try
-%         %% Destroy session if clean_start is not set.
-%         %% Ignore any crashes, just refresh the clean_start = true.
-%         _ = do_connect(State#{connect_opts => ConnectOpts#{clean_start => true}}),
-%         _ = disconnect(State),
-%         ok
-%     catch
-%         _:_ ->
-%             ok
-%     end;
-% maybe_destroy_session(_State) ->
-%     ok.