yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test(emqx_bridge_kafka_impl_consumer_SUITE): fix flaky

This commit is contained in:
Zaiming (Stone) Shi 2023-11-03 11:48:50 +01:00
parent afe6f79a66
commit 4dd054b0a2
1 changed files with 44 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
apps/emqx_bridge_kafka/test/emqx_bridge_kafka_impl_consumer_SUITE.erl
View File

@ -698,6 +698,20 @@ create_bridge(Config, Overrides) ->
KafkaConfig = emqx_utils_maps:deep_merge(KafkaConfig0, Overrides), KafkaConfig = emqx_utils_maps:deep_merge(KafkaConfig0, Overrides),
emqx_bridge:create(Type, Name, KafkaConfig). emqx_bridge:create(Type, Name, KafkaConfig).
create_bridge_wait_for_balance(Config) ->
setup_group_subscriber_spy(self()),
try
Res = create_bridge(Config),
receive
{kafka_assignment, _, _} ->
Res
after 20_000 ->
ct:fail("timed out waiting for kafka assignment")
end
after
kill_group_subscriber_spy()
end.
delete_bridge(Config) -> delete_bridge(Config) ->
Type = ?BRIDGE_TYPE_BIN, Type = ?BRIDGE_TYPE_BIN,
Name = ?config(kafka_name, Config), Name = ?config(kafka_name, Config),
@ -1020,6 +1034,10 @@ reconstruct_assignments_from_events(KafkaTopic, Events0, Acc0) ->
setup_group_subscriber_spy_fn() -> setup_group_subscriber_spy_fn() ->
TestPid = self(), TestPid = self(),
fun() -> fun() ->
setup_group_subscriber_spy(TestPid)
end.
setup_group_subscriber_spy(TestPid) ->
ok = meck:new(brod_group_subscriber_v2, [ ok = meck:new(brod_group_subscriber_v2, [
passthrough, no_link, no_history, non_strict passthrough, no_link, no_history, non_strict
]), ]),
@ -1042,8 +1060,10 @@ setup_group_subscriber_spy_fn() ->
meck:passthrough([Pid, MemberId, GenerationId, TopicAssignments]) meck:passthrough([Pid, MemberId, GenerationId, TopicAssignments])
end end
), ),
ok ok.
end.
kill_group_subscriber_spy() ->
meck:unload(brod_group_subscriber_v2).
wait_for_cluster_rpc(Node) -> wait_for_cluster_rpc(Node) ->
%% need to wait until the config handler is ready after %% need to wait until the config handler is ready after
@ -1702,10 +1722,7 @@ t_dynamic_mqtt_topic(Config) ->
MQTTTopic = emqx_topic:join([KafkaTopic, '#']), MQTTTopic = emqx_topic:join([KafkaTopic, '#']),
?check_trace( ?check_trace(
begin begin
?assertMatch( ?assertMatch({ok, _}, create_bridge_wait_for_balance(Config)),
{ok, _},
create_bridge(Config)
),
wait_until_subscribers_are_ready(NPartitions, 40_000), wait_until_subscribers_are_ready(NPartitions, 40_000),
{ok, C} = emqtt:start_link(), {ok, C} = emqtt:start_link(),
on_exit(fun() -> emqtt:stop(C) end), on_exit(fun() -> emqtt:stop(C) end),