test(emqx_resource): add regression test for recursive flushing

This commit is contained in:
Erik Timan 2023-02-16 14:14:29 +01:00
parent dcf70e0e68
commit 2442a4dea7
1 changed files with 61 additions and 0 deletions

View File

@ -2291,6 +2291,67 @@ t_expiration_retry_batch_multiple_times(_Config) ->
), ),
ok. ok.
t_recursive_flush(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
#{name => test_resource},
#{
query_mode => async,
batch_size => 1,
batch_time => 10_000,
worker_pool_size => 1
}
),
do_t_recursive_flush().
t_recursive_flush_batch(_Config) ->
emqx_connector_demo:set_callback_mode(async_if_possible),
{ok, _} = emqx_resource:create(
?ID,
?DEFAULT_RESOURCE_GROUP,
?TEST_RESOURCE,
#{name => test_resource},
#{
query_mode => async,
batch_size => 2,
batch_time => 10_000,
worker_pool_size => 1
}
),
do_t_recursive_flush().
do_t_recursive_flush() ->
?check_trace(
begin
Timeout = 1_000,
Pid = spawn_link(fun S() ->
emqx_resource:query(?ID, {inc_counter, 1}),
S()
end),
%% we want two reflushes to happen before we analyze the
%% trace, so that we get a single full interaction
{ok, _} = snabbkaffe:block_until(
?match_n_events(2, #{?snk_kind := buffer_worker_flush_ack_reflush}), Timeout
),
unlink(Pid),
exit(Pid, kill),
ok
end,
fun(Trace) ->
%% check that a recursive flush leads to a new call to flush/1
Pairs = ?find_pairs(
#{?snk_kind := buffer_worker_flush_ack_reflush},
#{?snk_kind := buffer_worker_flush},
Trace
),
?assert(lists:any(fun(E) -> E end, [true || {pair, _, _} <- Pairs]))
end
),
ok.
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% Helpers %% Helpers
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------