Merge pull request #9832 from sstrigler/EMQX-8774-failure-to-handle-timeout-error-in-resource-worker

EMQX 8774 failure to handle timeout error in resource worker

apps/emqx_bridge/test/emqx_bridge_mqtt_SUITE.erl

@@ -899,7 +899,7 @@ t_mqtt_conn_bridge_egress_reconnect(_) ->
         ),
     Payload1 = <<"hello2">>,
     Payload2 = <<"hello3">>,
-    %% we need to to it in other processes because it'll block due to
+    %% We need to do it in other processes because it'll block due to
     %% the long timeout
     spawn(fun() -> emqx:publish(emqx_message:make(LocalTopic, Payload1)) end),
     spawn(fun() -> emqx:publish(emqx_message:make(LocalTopic, Payload2)) end),

apps/emqx_resource/src/emqx_resource_buffer_worker.erl

@@ -20,7 +20,6 @@
 -module(emqx_resource_buffer_worker).
 
 -include("emqx_resource.hrl").
--include("emqx_resource_utils.hrl").
 -include("emqx_resource_errors.hrl").
 -include_lib("emqx/include/logger.hrl").
 -include_lib("stdlib/include/ms_transform.hrl").
@@ -266,15 +265,17 @@ code_change(_OldVsn, State, _Extra) ->
 
 %%==============================================================================
 -define(PICK(ID, KEY, PID, EXPR),
-    try gproc_pool:pick_worker(ID, KEY) of
-        PID when is_pid(PID) ->
-            EXPR;
-        _ ->
-            ?RESOURCE_ERROR(worker_not_created, "resource not created")
+    try
+        case gproc_pool:pick_worker(ID, KEY) of
+            PID when is_pid(PID) ->
+                EXPR;
+            _ ->
+                ?RESOURCE_ERROR(worker_not_created, "resource not created")
+        end
     catch
         error:badarg ->
             ?RESOURCE_ERROR(worker_not_created, "resource not created");
-        exit:{timeout, _} ->
+        error:timeout ->
             ?RESOURCE_ERROR(timeout, "call resource timeout")
     end
 ).

apps/emqx_resource/test/emqx_resource_SUITE.erl

@@ -19,8 +19,6 @@
 -compile(export_all).
 
 -include_lib("eunit/include/eunit.hrl").
--include_lib("common_test/include/ct.hrl").
--include("emqx_resource.hrl").
 -include_lib("stdlib/include/ms_transform.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
@@ -772,7 +770,10 @@ t_healthy_timeout(_) ->
         %% the ?TEST_RESOURCE always returns the `Mod:on_get_status/2` 300ms later.
         #{health_check_interval => 200}
     ),
-    ?assertError(timeout, emqx_resource:query(?ID, get_state, #{timeout => 1_000})),
+    ?assertMatch(
+        {error, {resource_error, #{reason := timeout}}},
+        emqx_resource:query(?ID, get_state, #{timeout => 1_000})
+    ),
     ?assertMatch({ok, _Group, #{status := disconnected}}, emqx_resource_manager:ets_lookup(?ID)),
     ok = emqx_resource:remove_local(?ID).
 
@@ -1583,8 +1584,8 @@ do_t_expiration_before_sending(QueryMode) ->
             spawn_link(fun() ->
                 case QueryMode of
                     sync ->
-                        ?assertError(
-                            timeout,
+                        ?assertMatch(
+                            {error, {resource_error, #{reason := timeout}}},
                             emqx_resource:query(?ID, {inc_counter, 99}, #{timeout => TimeoutMS})
                         );
                     async ->
@@ -1690,8 +1691,8 @@ do_t_expiration_before_sending_partial_batch(QueryMode) ->
                 spawn_link(fun() ->
                     case QueryMode of
                         sync ->
-                            ?assertError(
-                                timeout,
+                            ?assertMatch(
+                                {error, {resource_error, #{reason := timeout}}},
                                 emqx_resource:query(?ID, {inc_counter, 199}, #{timeout => TimeoutMS})
                             );
                         async ->
@@ -2043,8 +2044,8 @@ do_t_expiration_retry(IsBatch) ->
                 ResumeInterval * 2
             ),
             spawn_link(fun() ->
-                ?assertError(
-                    timeout,
+                ?assertMatch(
+                    {error, {resource_error, #{reason := timeout}}},
                     emqx_resource:query(
                         ?ID,
                         {inc_counter, 1},
@@ -2127,8 +2128,8 @@ t_expiration_retry_batch_multiple_times(_Config) ->
             ),
             TimeoutMS = 100,
             spawn_link(fun() ->
-                ?assertError(
-                    timeout,
+                ?assertMatch(
+                    {error, {resource_error, #{reason := timeout}}},
                     emqx_resource:query(
                         ?ID,
                         {inc_counter, 1},
@@ -2137,8 +2138,8 @@ t_expiration_retry_batch_multiple_times(_Config) ->
                 )
             end),
             spawn_link(fun() ->
-                ?assertError(
-                    timeout,
+                ?assertMatch(
+                    {error, {resource_error, #{reason := timeout}}},
                     emqx_resource:query(
                         ?ID,
                         {inc_counter, 2},

changes/v5.0.16/fix-9832.en.md

@@ -0,0 +1 @@
+Improve error log when bridge in 'sync' mode timed out to get response.

changes/v5.0.16/fix-9832.zh.md

@@ -0,0 +1 @@
+优化桥接同步资源调用超时情况下的一个错误日志。

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_influxdb_SUITE.erl

@@ -910,12 +910,10 @@ t_write_failure(Config) ->
                 sync ->
                     {_, {ok, _}} =
                         ?wait_async_action(
-                            try
+                            ?assertMatch(
+                                {error, {resource_error, #{reason := timeout}}},
                                 send_message(Config, SentData)
-                            catch
-                                error:timeout ->
-                                    {error, timeout}
-                            end,
+                            ),
                             #{?snk_kind := buffer_worker_flush_nack},
                             1_000
                         );
@@ -947,7 +945,8 @@ t_write_failure(Config) ->
                         {error, {recoverable_error, {closed, "The connection was lost."}}} =:=
                             Result orelse
                             {error, {error, closed}} =:= Result orelse
-                            {error, {recoverable_error, econnrefused}} =:= Result,
+                            {error, {recoverable_error, econnrefused}} =:= Result orelse
+                            {error, {recoverable_error, noproc}} =:= Result,
                         #{got => Result}
                     )
             end,

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_mysql_SUITE.erl

@@ -406,7 +406,10 @@ t_write_failure(Config) ->
         emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
             case QueryMode of
                 sync ->
-                    ?assertError(timeout, send_message(Config, SentData));
+                    ?assertMatch(
+                        {error, {resource_error, #{reason := timeout}}},
+                        send_message(Config, SentData)
+                    );
                 async ->
                     send_message(Config, SentData)
             end
@@ -439,8 +442,8 @@ t_write_timeout(Config) ->
     SentData = #{payload => Val, timestamp => 1668602148000},
     Timeout = 1000,
     emqx_common_test_helpers:with_failure(timeout, ProxyName, ProxyHost, ProxyPort, fun() ->
-        ?assertError(
-            timeout,
+        ?assertMatch(
+            {error, {resource_error, #{reason := timeout}}},
             query_resource(Config, {send_message, SentData, [], Timeout})
         )
     end),

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_pgsql_SUITE.erl

@@ -426,12 +426,7 @@ t_write_failure(Config) ->
                 ?wait_async_action(
                     case QueryMode of
                         sync ->
-                            try
-                                send_message(Config, SentData)
-                            catch
-                                error:timeout ->
-                                    {error, timeout}
-                            end;
+                            ?assertMatch({error, _}, send_message(Config, SentData));
                         async ->
                             send_message(Config, SentData)
                     end,
@@ -467,7 +462,10 @@ t_write_timeout(Config) ->
     SentData = #{payload => Val, timestamp => 1668602148000},
     Timeout = 1000,
     emqx_common_test_helpers:with_failure(timeout, ProxyName, ProxyHost, ProxyPort, fun() ->
-        ?assertError(timeout, query_resource(Config, {send_message, SentData, [], Timeout}))
+        ?assertMatch(
+            {error, {resource_error, #{reason := timeout}}},
+            query_resource(Config, {send_message, SentData, [], Timeout})
+        )
     end),
     ok.