Merge pull request #12282 from thalesmg/fix-mysql-del-action-m-20240109

fix(mysql_bridge): allow deleting bridge when sql contains undefined fields

apps/emqx_bridge/test/emqx_bridge_testlib.erl

@@ -184,6 +184,15 @@ update_bridge_api(Config, Overrides) ->
     ct:pal("bridge update result: ~p", [Res]),
     Res.
 
+delete_bridge_http_api_v1(Opts) ->
+    #{type := Type, name := Name} = Opts,
+    BridgeId = emqx_bridge_resource:bridge_id(Type, Name),
+    Path = emqx_mgmt_api_test_util:api_path(["bridges", BridgeId]),
+    ct:pal("deleting bridge (http v1)"),
+    Res = emqx_bridge_v2_testlib:request(delete, Path, _Params = []),
+    ct:pal("bridge delete (http v1) result:\n  ~p", [Res]),
+    Res.
+
 op_bridge_api(Op, BridgeType, BridgeName) ->
     BridgeId = emqx_bridge_resource:bridge_id(BridgeType, BridgeName),
     Path = emqx_mgmt_api_test_util:api_path(["bridges", BridgeId, Op]),

apps/emqx_bridge_mysql/src/emqx_bridge_mysql.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_mysql, [
     {description, "EMQX Enterprise MySQL Bridge"},
-    {vsn, "0.1.3"},
+    {vsn, "0.1.4"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_bridge_mysql/src/emqx_bridge_mysql_connector.erl

@@ -39,11 +39,23 @@ on_add_channel(
         ok ->
             ChannelConfig2 = maps:merge(ChannelConfig1, QueryTemplates),
             ChannelConfig = set_prepares(ChannelConfig2, ConnectorState),
-            State = State0#{
-                channels => maps:put(ChannelId, ChannelConfig, Channels),
-                connector_state => ConnectorState
-            },
-            {ok, State};
+            case maps:get(prepares, ChannelConfig) of
+                {error, {Code, ErrState, Msg}} ->
+                    Context = #{
+                        code => Code,
+                        state => ErrState,
+                        message => Msg
+                    },
+                    {error, {prepare_statement, Context}};
+                {error, undefined_table} ->
+                    {error, {unhealthy_target, <<"Undefined table">>}};
+                _ ->
+                    State = State0#{
+                        channels => maps:put(ChannelId, ChannelConfig, Channels),
+                        connector_state => ConnectorState
+                    },
+                    {ok, State}
+            end;
         {error, Error} ->
             {error, Error}
     end.

apps/emqx_bridge_mysql/test/emqx_bridge_mysql_SUITE.erl

@@ -50,7 +50,9 @@ groups() ->
     NonBatchCases = [
         t_write_timeout,
         t_uninitialized_prepared_statement,
-        t_non_batch_update_is_allowed
+        t_non_batch_update_is_allowed,
+        t_delete_with_undefined_field_in_sql,
+        t_undefined_field_in_sql
     ],
     OnlyBatchCases = [
         t_batch_update_is_forbidden
@@ -801,27 +803,13 @@ t_missing_table(Config) ->
                     sync ->
                         query_resource(Config, Request);
                     async ->
-                        {_, Ref} = query_resource_async(Config, Request),
-                        {ok, Res} = receive_result(Ref, 2_000),
+                        {Res, _Ref} = query_resource_async(Config, Request),
                         Res
                 end,
-
-            BatchSize = ?config(batch_size, Config),
-            IsBatch = BatchSize > 1,
-            case IsBatch of
-                true ->
-                    ?assertMatch(
-                        {error,
-                            {unrecoverable_error,
-                                {1146, <<"42S02">>, <<"Table 'mqtt.mqtt_test' doesn't exist">>}}},
-                        Result
-                    );
-                false ->
-                    ?assertMatch(
-                        {error, undefined_table},
-                        Result
-                    )
-            end,
+            ?assertMatch(
+                {error, {resource_error, #{reason := unhealthy_target}}},
+                Result
+            ),
             ok
         end,
         fun(Trace) ->
@@ -974,3 +962,58 @@ t_non_batch_update_is_allowed(Config) ->
         []
     ),
     ok.
+
+t_undefined_field_in_sql(Config) ->
+    ?check_trace(
+        begin
+            Overrides = #{
+                <<"sql">> =>
+                    <<
+                        "INSERT INTO mqtt_test(wrong_column, arrived) "
+                        "VALUES (${payload}, FROM_UNIXTIME(${timestamp}/1000))"
+                    >>
+            },
+            ProbeRes = emqx_bridge_testlib:probe_bridge_api(Config, Overrides),
+            ?assertMatch({error, {{_, 400, _}, _, _BodyRaw}}, ProbeRes),
+            {error, {{_, 400, _}, _, BodyRaw}} = ProbeRes,
+            ?assertEqual(
+                match,
+                re:run(
+                    BodyRaw,
+                    <<"Unknown column 'wrong_column' in 'field list'">>,
+                    [{capture, none}]
+                ),
+                #{body => BodyRaw}
+            ),
+            ok
+        end,
+        []
+    ),
+    ok.
+
+t_delete_with_undefined_field_in_sql(Config) ->
+    ?check_trace(
+        begin
+            Name = ?config(bridge_name, Config),
+            Type = ?config(bridge_type, Config),
+            Overrides = #{
+                <<"sql">> =>
+                    <<
+                        "INSERT INTO mqtt_test(wrong_column, arrived) "
+                        "VALUES (${payload}, FROM_UNIXTIME(${timestamp}/1000))"
+                    >>
+            },
+            ?assertMatch(
+                {ok, {{_, 201, _}, _, #{<<"status">> := Status}}} when
+                    Status =:= <<"connecting">> orelse Status =:= <<"disconnected">>,
+                emqx_bridge_testlib:create_bridge_api(Config, Overrides)
+            ),
+            ?assertMatch(
+                {ok, {{_, 204, _}, _, _}},
+                emqx_bridge_testlib:delete_bridge_http_api_v1(#{type => Type, name => Name})
+            ),
+            ok
+        end,
+        []
+    ),
+    ok.

changes/ee/fix-12282.en.md

@@ -0,0 +1,3 @@
+Improved HTTP API error message when the creation of a MySQL bridge fails.
+
+Fixed an issue that prevented removing a MySQL bridge when its SQL contained undefined columns.