chore: fix flaky test cases, log labels and review comments

Co-authored-by: Thales Macedo Garitezi <thalesmg@gmail.com>

apps/emqx_resource/src/emqx_resource_manager.erl

@@ -428,14 +428,6 @@ read_cache(ResId) ->
         [] -> not_found
     end.
 
-read_manager_pid_from_cache(ResId) ->
-    case read_cache(ResId) of
-        not_found ->
-            erlang:error(badarg);
-        {_, #data{pid = ManagerPid}} ->
-            ManagerPid
-    end.
-
 delete_cache(ResId, MgrId) ->
     case get_owner(ResId) of
         MgrIdNow when MgrIdNow == not_found; MgrIdNow == MgrId ->
@@ -656,8 +648,12 @@ do_wait_for_ready(ResId, Retry) ->
 
 safe_call(ResId, Message, Timeout) ->
     try
-        ManagerPid = read_manager_pid_from_cache(ResId),
+        case read_cache(ResId) of
+            not_found ->
+                {error, not_found};
+            {_, #data{pid = ManagerPid}} ->
                 gen_statem:call(ManagerPid, Message, {clean_timeout, Timeout})
+        end
     catch
         error:badarg ->
             {error, not_found};

apps/emqx_resource/src/emqx_resource_worker.erl

@@ -164,7 +164,7 @@ init({Id, Index, Opts}) ->
     emqx_resource_metrics:queuing_set(Id, Index, queue_count(Queue)),
     emqx_resource_metrics:inflight_set(Id, Index, 0),
     InfltWinSZ = maps:get(async_inflight_window, Opts, ?DEFAULT_INFLIGHT),
-    {ok, InflightTID} = inflight_new(InfltWinSZ, Id, Index),
+    InflightTID = inflight_new(InfltWinSZ, Id, Index),
     HCItvl = maps:get(health_check_interval, Opts, ?HEALTHCHECK_INTERVAL),
     St = #{
         id => Id,
@@ -765,7 +765,7 @@ inflight_new(InfltWinSZ, Id, Index) ->
     %% we use this counter because we might deal with batches as
     %% elements.
     inflight_append(TableId, ?SIZE_REF, 0, Id, Index),
-    {ok, TableId}.
+    TableId.
 
 inflight_get_first(InflightTID) ->
     case ets:next(InflightTID, ?MAX_SIZE_REF) of

apps/emqx_resource/test/emqx_resource_SUITE.erl

@@ -944,7 +944,15 @@ t_create_dry_run_local(_) ->
         end,
         lists:seq(1, 10)
     ),
-    [] = ets:match(emqx_resource_manager, {{owner, '$1'}, '_'}).
+    case [] =:= ets:match(emqx_resource_manager, {{owner, '$1'}, '_'}) of
+        false ->
+            %% Sleep to remove flakyness in test case. It take some time for
+            %% the ETS table to be cleared.
+            timer:sleep(2000),
+            [] = ets:match(emqx_resource_manager, {{owner, '$1'}, '_'});
+        true ->
+            ok
+    end.
 
 create_dry_run_local_succ() ->
     case whereis(test_resource) of