Merge pull request #9638 from lafirest/fix/mysql_dup

fix(mysql): fix the problem of data loss and bad match when mysql is disconnected

apps/emqx_connector/src/emqx_connector_mysql.erl

@@ -408,20 +408,34 @@ on_sql_query(
     LogMeta = #{connector => InstId, sql => SQLOrKey, state => State},
     ?TRACE("QUERY", "mysql_connector_received", LogMeta),
     Worker = ecpool:get_client(PoolName),
-    {ok, Conn} = ecpool_worker:client(Worker),
+    case ecpool_worker:client(Worker) of
-    ?tp(
+        {ok, Conn} ->
-        mysql_connector_send_query,
+            ?tp(
-        #{sql_func => SQLFunc, sql_or_key => SQLOrKey, data => Data}
+                mysql_connector_send_query,
-    ),
+                #{sql_func => SQLFunc, sql_or_key => SQLOrKey, data => Data}
+            ),
+            do_sql_query(SQLFunc, Conn, SQLOrKey, Data, Timeout, LogMeta);
+        {error, disconnected} ->
+            ?SLOG(
+                error,
+                LogMeta#{
+                    msg => "mysql_connector_do_sql_query_failed",
+                    reason => worker_is_disconnected
+                }
+            ),
+            {error, {recoverable_error, disconnected}}
+    end.
+
+do_sql_query(SQLFunc, Conn, SQLOrKey, Data, Timeout, LogMeta) ->
     try mysql:SQLFunc(Conn, SQLOrKey, Data, Timeout) of
-        {error, disconnected} = Result ->
+        {error, disconnected} ->
             ?SLOG(
                 error,
                 LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => disconnected}
             ),
             %% kill the poll worker to trigger reconnection
             _ = exit(Conn, restart),
-            Result;
+            {error, {recoverable_error, disconnected}};
         {error, not_prepared} = Error ->
             ?tp(
                 mysql_connector_prepare_query_failed,

changes/v5.0.14-en.md

@@ -10,3 +10,5 @@
 ## Bug Fixes
 
 - Fix an issue where testing the GCP PubSub could leak memory, and an issue where its JWT token would fail to refresh a second time. [#9641](https://github.com/emqx/emqx/pull/9641)
+
+- Fix the problem of data loss and bad match when the MySQL driver is disconnected [#9638](https://github.com/emqx/emqx/pull/9638).

changes/v5.0.14-zh.md

@@ -10,3 +10,5 @@
 ## 修复
 
 - 修复了测试GCP PubSub可能泄露内存的问题，以及其JWT令牌第二次刷新失败的问题。 [#9640](https://github.com/emqx/emqx/pull/9640)
+
+- 修复 MySQL 驱动断开连接时出现的数据丢失和匹配错误的问题 [#9638](https://github.com/emqx/emqx/pull/9638)。

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl

@@ -404,9 +404,13 @@ t_write_failure(Config) ->
         emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
             send_message(Config, SentData)
         end),
-        fun(Result, _Trace) ->
+        fun
-            ?assertMatch({error, {resource_error, _}}, Result),
+            ({error, {resource_error, _}}, _Trace) ->
-            ok
+                ok;
+            ({error, {recoverable_error, disconnected}}, _Trace) ->
+                ok;
+            (_, _Trace) ->
+                ?assert(false)
         end
     ),
     ok.