From bf3983e7c46e54a7e68ba713291624f35a430c1a Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi Date: Tue, 27 Dec 2022 11:23:13 -0300 Subject: [PATCH] feat(buffer_worker): use offload mode for `replayq` To avoid confusion for the users as to what persistence guarantees we offer when buffering bridges/resources, we will always enable offload mode for `replayq`. With this, when the buffer size is above the max segment size, it'll flush the queue to disk, but on recovery after a restart it'll clean the existing segments rather than resuming from them. --- apps/emqx_resource/src/emqx_resource_worker.erl | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/apps/emqx_resource/src/emqx_resource_worker.erl b/apps/emqx_resource/src/emqx_resource_worker.erl index 5f7cdf7e0..f04f7b1bd 100644 --- a/apps/emqx_resource/src/emqx_resource_worker.erl +++ b/apps/emqx_resource/src/emqx_resource_worker.erl @@ -139,10 +139,13 @@ init({Id, Index, Opts}) -> true -> replayq:open(#{ dir => disk_queue_dir(Id, Index), - seg_bytes => SegBytes, + marshaller => fun ?MODULE:queue_item_marshaller/1, max_total_bytes => TotalBytes, - sizer => fun ?MODULE:estimate_size/1, - marshaller => fun ?MODULE:queue_item_marshaller/1 + %% we don't want to retain the queue after + %% resource restarts. + offload => true, + seg_bytes => SegBytes, + sizer => fun ?MODULE:estimate_size/1 }); false -> undefined