fix(mysql_bridge): check prepare statement error messages

Fixes https://emqx.atlassian.net/browse/EMQX-11648

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),
+            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};
+                    {ok, State}
+            end;
         {error, Error} ->
             {error, Error}
     end.

apps/emqx_bridge_mysql/test/emqx_bridge_mysql_SUITE.erl

@@ -50,7 +50,8 @@ groups() ->
     NonBatchCases = [
         t_write_timeout,
         t_uninitialized_prepared_statement,
-        t_non_batch_update_is_allowed
+        t_non_batch_update_is_allowed,
+        t_undefined_field_in_sql
     ],
     OnlyBatchCases = [
         t_batch_update_is_forbidden
@@ -801,27 +802,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">>}}},
+                {error, {resource_error, #{reason := unhealthy_target}}},
                 Result
-                    );
-                false ->
-                    ?assertMatch(
-                        {error, undefined_table},
-                        Result
-                    )
-            end,
+            ),
             ok
         end,
         fun(Trace) ->
@@ -974,3 +961,31 @@ 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.

changes/ee/fix-12281.en.md

@@ -0,0 +1 @@
+Improved HTTP API error message when the creation of a MySQL bridge fails.