From 9d20431257a45af08e0ed4e43df1c4675d00b5b1 Mon Sep 17 00:00:00 2001
From: Erik Timan <erik.timan@emqx.io>
Date: Tue, 24 Jan 2023 14:13:35 +0100
Subject: [PATCH] fix(emqx_resource): fix crash while flushing queue

We used next_event for flushing the queue in emqx_resource, but this
leads to a crash. We now call flush_worker/1 instead.
---
 .../src/emqx_resource_buffer_worker.erl       | 27 ++++++++++---------
 1 file changed, 15 insertions(+), 12 deletions(-)

diff --git a/apps/emqx_resource/src/emqx_resource_buffer_worker.erl b/apps/emqx_resource/src/emqx_resource_buffer_worker.erl
index 669b8e474..634b0e954 100644
--- a/apps/emqx_resource/src/emqx_resource_buffer_worker.erl
+++ b/apps/emqx_resource/src/emqx_resource_buffer_worker.erl
@@ -583,10 +583,11 @@ do_flush(
             ),
             case queue_count(Q1) > 0 of
                 true ->
-                    {keep_state, Data1, [{next_event, internal, flush}]};
+                    flush_worker(self());
                 false ->
-                    {keep_state, Data1}
-            end
+                    ok
+            end,
+            {keep_state, Data1}
     end;
 do_flush(Data0, #{
     is_batch := true,
@@ -659,15 +660,17 @@ do_flush(Data0, #{
                 }
             ),
             CurrentCount = queue_count(Q1),
-            case {CurrentCount > 0, CurrentCount >= BatchSize} of
-                {false, _} ->
-                    {keep_state, Data1};
-                {true, true} ->
-                    {keep_state, Data1, [{next_event, internal, flush}]};
-                {true, false} ->
-                    Data2 = ensure_flush_timer(Data1),
-                    {keep_state, Data2}
-            end
+            Data2 =
+                case {CurrentCount > 0, CurrentCount >= BatchSize} of
+                    {false, _} ->
+                        Data1;
+                    {true, true} ->
+                        flush_worker(self()),
+                        Data1;
+                    {true, false} ->
+                        ensure_flush_timer(Data1)
+                end,
+            {keep_state, Data2}
     end.
 
 batch_reply_caller(Id, BatchResult, Batch, QueryOpts) ->