Skip to content

Commit c117a35

Browse files
zmstonezmstone
committed
fix(wolff_client): do not receive shutdown EXIT signal
1 parent 127c25a commit c117a35

2 files changed

Lines changed: 54 additions & 51 deletions

File tree

‎src/wolff_client.erl‎

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,8 @@
5757
%% but we need to find connection by {topic(), partition()}
5858
leaders => #{{topic(), partition()} => connection()},
5959
%% Reference counting so we may drop connection metadata when no longer required.
60-
known_topics := #{topic() => #{producer_group() => true}}
60+
known_topics := #{topic() => #{producer_group() => true}},
61+
owner := pid()
6162
}.
6263

6364
-define(DEFAULT_METADATA_TIMEOUT, 10000).
@@ -78,6 +79,7 @@
7879
start_link(ClientId, Hosts, Config) ->
7980
{ConnCfg0, MyCfg} = split_config(Config),
8081
ConnCfg = ConnCfg0#{client_id => ClientId},
82+
Owner = self(),
8183
St = #{client_id => ClientId,
8284
seed_hosts => Hosts,
8385
config => MyCfg,
@@ -86,7 +88,8 @@ start_link(ClientId, Hosts, Config) ->
8688
metadata_conn => not_initialized,
8789
metadata_ts => #{},
8890
leaders => #{},
89-
known_topics => #{}
91+
known_topics => #{},
92+
owner => Owner
9093
},
9194
case maps:get(reg_name, Config, false) of
9295
false -> gen_server:start_link(?MODULE, St, []);
@@ -199,7 +202,13 @@ handle_call(Call, _From, St) ->
199202
{reply, {error, {unknown_call, Call}}, St}.
200203

201204
handle_info({'EXIT', Pid, Reason}, St) ->
202-
{noreply, flush_exit_signals(St, Pid, Reason)};
205+
%% Check if this is a shutdown signal from the supervisor
206+
case erlang:whereis(wolff_client_sup) of
207+
Pid when Reason =:= shutdown ->
208+
{stop, shutdown, St};
209+
_ ->
210+
{noreply, flush_exit_signals(St, Pid, Reason)}
211+
end;
203212
handle_info(_Info, St) ->
204213
{noreply, upgrade(St)}.
205214

@@ -243,7 +252,7 @@ handle_cast(_Cast, St) ->
243252
code_change(_OldVsn, St, _Extra) ->
244253
{ok, St}.
245254

246-
terminate(_, #{client_id := ClientID, conns := Conns} = St) ->
255+
terminate(Reason, #{client_id := ClientID, conns := Conns} = St) ->
247256
ok = wolff_client_sup:deregister_client(ClientID),
248257
MetadataConn = maps:get(metadata_conn, St, none),
249258
ok = close_connections(Conns),
@@ -307,9 +316,9 @@ close_connections(Conns, Topic) ->
307316
end,
308317
do_close_connections(maps:to_list(Conns), Pred, #{}).
309318

310-
flush_exit_signals(St0) ->
319+
flush_exit_signals(#{owner := Owner} = St0) ->
311320
receive
312-
{'EXIT', Pid, Reason} ->
321+
{'EXIT', Pid, Reason} when Pid =/= Owner ->
313322
flush_exit_signals(St0, Pid, Reason)
314323
after
315324
0 ->

‎test/wolff_supervised_tests.erl‎

Lines changed: 39 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,7 @@ supervised_client_test() ->
1111
CntrEventsTable = ets:new(cntr_events, [public]),
1212
wolff_tests:install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),
1313
ClientId = <<"supervised-wolff-client">>,
14-
_ = application:stop(wolff), %% ensure stopped
15-
{ok, _} = application:ensure_all_started(wolff),
14+
ok = start_app(),
1615
ClientCfg = client_config(),
1716
{ok, Client} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
1817
%% start it again should result in the same client pid
@@ -28,14 +27,15 @@ supervised_client_test() ->
2827
ok = wolff:stop_producers(Producers),
2928
ok = wolff:stop_and_delete_supervised_client(ClientId),
3029
?assertEqual([], supervisor:which_children(wolff_client_sup)),
31-
ok = application:stop(wolff),
30+
ok = stop_app(),
3231
?assertEqual(undefined, whereis(wolff_sup)),
3332
assert_last_event_is_zero(queuing, CntrEventsTable),
3433
assert_last_event_is_zero(queuing_bytes, CntrEventsTable),
3534
assert_last_event_is_zero(inflight, CntrEventsTable),
3635
[1] = get_telemetry_seq(CntrEventsTable, [wolff,success]),
3736
ets:delete(CntrEventsTable),
3837
wolff_tests:deinstall_event_logging(?FUNCTION_NAME),
38+
stop_app(),
3939
ok.
4040

4141
supervised_producers_test_() ->
@@ -46,9 +46,8 @@ supervised_producers_test_() ->
4646
test_supervised_producers(Name) ->
4747
CntrEventsTable = ets:new(cntr_events, [public]),
4848
wolff_tests:install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),
49-
ClientId = <<"supervised-producers">>,
50-
_ = application:stop(wolff), %% ensure stopped
51-
{ok, _} = application:ensure_all_started(wolff),
49+
ClientId = <<"supervised-producers-1">>,
50+
ok = start_app(),
5251
ClientCfg = client_config(),
5352
{ok, _ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
5453
ProducerCfg0 = producer_config(Name),
@@ -64,7 +63,7 @@ test_supervised_producers(Name) ->
6463
?assertEqual([], supervisor:which_children(wolff_producers_sup)),
6564
ok = wolff:stop_and_delete_supervised_client(ClientId),
6665
?assertEqual([], supervisor:which_children(wolff_client_sup)),
67-
ok = application:stop(wolff),
66+
ok = stop_app(),
6867
assert_last_event_is_zero(queuing, CntrEventsTable),
6968
assert_last_event_is_zero(queuing_bytes, CntrEventsTable),
7069
assert_last_event_is_zero(inflight, CntrEventsTable),
@@ -78,8 +77,7 @@ different_producers_same_topic_test_() ->
7877
{timeout, 30, fun test_different_producers_same_topic/0}.
7978

8079
test_different_producers_same_topic() ->
81-
_ = application:stop(wolff), %% ensure stopped
82-
{ok, _} = application:ensure_all_started(wolff),
80+
ok = start_app(),
8381
ClientId = <<"same-topic">>,
8482
ClientCfg = client_config(),
8583
{ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
@@ -134,7 +132,7 @@ test_different_producers_same_topic() ->
134132
},
135133
sys:get_state(ClientPid)),
136134
ok = wolff:stop_and_delete_supervised_client(ClientId),
137-
ok = application:stop(wolff),
135+
ok = stop_app(),
138136
ok.
139137

140138
client_restart_test() ->
@@ -153,8 +151,7 @@ client_restart_2_test() ->
153151
test_client_restart(ClientId, Topic, Partition) ->
154152
CntrEventsTable = ets:new(cntr_events, [public]),
155153
wolff_tests:install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),
156-
_ = application:stop(wolff), %% ensure stopped
157-
{ok, _} = application:ensure_all_started(wolff),
154+
ok = start_app(),
158155
ClientCfg = #{connection_strategy => per_broker},
159156
{ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
160157
ProducerCfg = #{replayq_dir => "test-data/client-restart-test",
@@ -177,7 +174,7 @@ test_client_restart(ClientId, Topic, Partition) ->
177174
?assertEqual([], supervisor:which_children(wolff_producers_sup)),
178175
ok = wolff:stop_and_delete_supervised_client(ClientId),
179176
?assertEqual([], supervisor:which_children(wolff_client_sup)),
180-
ok = application:stop(wolff),
177+
ok = stop_app(),
181178
[1,1] = get_telemetry_seq(CntrEventsTable, [wolff,success]),
182179
assert_last_event_is_zero(queuing, CntrEventsTable),
183180
assert_last_event_is_zero(queuing_bytes, CntrEventsTable),
@@ -191,8 +188,7 @@ max_partitions_test() ->
191188
Topic = <<"test-topic-2">>,
192189
Partition = 0,
193190
MaxPartitions = 1,
194-
_ = application:stop(wolff), %% ensure stopped
195-
{ok, _} = application:ensure_all_started(wolff),
191+
ok = start_app(),
196192
ClientCfg = #{connection_strategy => per_partition},
197193
{ok, _ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
198194
ProducerCfg = #{required_acks => all_isr,
@@ -207,16 +203,15 @@ max_partitions_test() ->
207203
?assertEqual([], supervisor:which_children(wolff_producers_sup)),
208204
ok = wolff:stop_and_delete_supervised_client(ClientId),
209205
?assertEqual([], supervisor:which_children(wolff_client_sup)),
210-
ok = application:stop(wolff),
206+
ok = stop_app(),
211207
ok.
212208

213209
%% Test against a bad host.
214210
%% No connection will be established at all.
215211
%% Producer workers should not crash, async APIs should work.
216212
bad_host_test() ->
217213
ClientId = <<"bad-host-test">>,
218-
_ = application:stop(wolff), %% ensure stopped
219-
{ok, _} = application:ensure_all_started(wolff),
214+
ok = start_app(),
220215
{ok, _} = wolff:ensure_supervised_client(ClientId, [{"badhost", 9092}], #{}),
221216
?assertMatch({error, _}, wolff:ensure_supervised_producers(ClientId, <<"t">>, #{name => ?FUNCTION_NAME})),
222217
ok = wolff:stop_and_delete_supervised_client(ClientId).
@@ -227,8 +222,7 @@ producer_restart_test() ->
227222
ClientId = <<"producer-restart">>,
228223
Topic = <<"test-topic">>,
229224
Partition = 0,
230-
_ = application:stop(wolff), %% ensure stopped
231-
{ok, _} = application:ensure_all_started(wolff),
225+
ok = start_app(),
232226
ClientCfg = #{},
233227
{ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
234228
ProducerCfg = #{replayq_dir => "test-data/producer-restart-test",
@@ -259,7 +253,7 @@ producer_restart_test() ->
259253
?assertEqual([], supervisor:which_children(wolff_producers_sup)),
260254
ok = wolff:stop_and_delete_supervised_client(ClientId),
261255
?assertEqual([], supervisor:which_children(wolff_client_sup)),
262-
ok = application:stop(wolff),
256+
ok = stop_app(),
263257
[1,2] = get_telemetry_seq(CntrEventsTable, [wolff,success]),
264258
assert_last_event_is_zero(queuing, CntrEventsTable),
265259
assert_last_event_is_zero(queuing_bytes, CntrEventsTable),
@@ -272,8 +266,7 @@ stop_with_name_test() ->
272266
ClientId = <<"stop-with-name">>,
273267
Topic = <<"test-topic">>,
274268
Partition = 0,
275-
_ = application:stop(wolff), %% ensure stopped
276-
{ok, _} = application:ensure_all_started(wolff),
269+
ok = start_app(),
277270
ClientCfg = #{},
278271
{ok, _} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
279272
Name = ?FUNCTION_NAME,
@@ -288,7 +281,7 @@ stop_with_name_test() ->
288281
?assertEqual([], supervisor:which_children(wolff_producers_sup)),
289282
ok = wolff:stop_and_delete_supervised_client(ClientId),
290283
?assertEqual([], supervisor:which_children(wolff_client_sup)),
291-
ok = application:stop(wolff),
284+
ok = stop_app(),
292285
ok.
293286

294287
partition_count_increase_test_() ->
@@ -300,8 +293,7 @@ test_partition_count_increase() ->
300293
wolff_tests:install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),
301294
ClientId = <<"test-add-more-partitions">>,
302295
Topic = <<"test-topic-3">>,
303-
_ = application:stop(wolff), %% ensure stopped
304-
{ok, _} = application:ensure_all_started(wolff),
296+
ok = start_app(),
305297
ClientCfg = #{},
306298
{ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
307299
{ok, Connections0} = get_leader_connections(ClientPid, Topic),
@@ -349,8 +341,7 @@ test_partition_count_decrease() ->
349341
Partitions0 = 3,
350342
delete_topic(Topic),
351343
create_topic(Topic, Partitions0),
352-
_ = application:stop(wolff), %% ensure stopped
353-
{ok, _} = application:ensure_all_started(wolff),
344+
ok = start_app(),
354345
ClientCfg = #{},
355346
{ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
356347
{ok, Connections0} = get_leader_connections(ClientPid, Topic),
@@ -409,7 +400,7 @@ test_partition_count_decrease() ->
409400
?assertEqual([], supervisor:which_children(wolff_producers_sup)),
410401
ok = wolff:stop_and_delete_supervised_client(ClientId),
411402
?assertEqual([], supervisor:which_children(wolff_client_sup)),
412-
ok = application:stop(wolff),
403+
ok = stop_app(),
413404
ok.
414405

415406
%% The last partition is temporarily missing from metadata response.
@@ -432,8 +423,7 @@ test_partition_missing_in_metadata_response(ThePartition) ->
432423
Partitions = 3,
433424
create_topic(Topic, Partitions),
434425
io:format(user, "created topic ~s with ~p partitions\n", [Topic, Partitions]),
435-
_ = application:stop(wolff), %% ensure stopped
436-
{ok, _} = application:ensure_all_started(wolff),
426+
ok = start_app(),
437427
%% always refresh metadata
438428
ClientCfg = #{min_metadata_refresh_interval => 0},
439429
{ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
@@ -501,7 +491,7 @@ test_partition_missing_in_metadata_response(ThePartition) ->
501491
?assertEqual([], supervisor:which_children(wolff_producers_sup)),
502492
ok = wolff:stop_and_delete_supervised_client(ClientId),
503493
?assertEqual([], supervisor:which_children(wolff_client_sup)),
504-
ok = application:stop(wolff),
494+
ok = stop_app(),
505495
ok.
506496

507497
get_partition_leader_connection(Client, Topic, Partition) ->
@@ -525,8 +515,7 @@ test_topic_recreate() ->
525515
delete_topic(Topic),
526516
Partitions0 = 3,
527517
create_topic(Topic, Partitions0),
528-
_ = application:stop(wolff), %% ensure stopped
529-
{ok, _} = application:ensure_all_started(wolff),
518+
ok = start_app(),
530519
ClientCfg = #{min_metadata_refresh_interval => 0},
531520
{ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
532521
{ok, Connections0} = get_leader_connections(ClientPid, Topic),
@@ -572,14 +561,13 @@ test_topic_recreate() ->
572561
?assertEqual([], supervisor:which_children(wolff_producers_sup)),
573562
ok = wolff:stop_and_delete_supervised_client(ClientId),
574563
?assertEqual([], supervisor:which_children(wolff_client_sup)),
575-
ok = application:stop(wolff),
564+
ok = stop_app(),
576565
ok.
577566

578567
non_existing_topic_test() ->
579568
ClientId = atom_to_binary(?FUNCTION_NAME),
580569
Topic = <<"non-existing-topic">>,
581-
_ = application:stop(wolff), %% ensure stopped
582-
{ok, _} = application:ensure_all_started(wolff),
570+
ok = start_app(),
583571
ClientCfg = #{},
584572
{ok, _ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
585573
?assertMatch({error, _}, wolff:ensure_supervised_producers(ClientId, Topic, #{name => ?FUNCTION_NAME})),
@@ -590,8 +578,7 @@ non_existing_topic_test() ->
590578
start_producers_with_dead_client_test() ->
591579
ClientId = atom_to_binary(?FUNCTION_NAME),
592580
Topic = <<"non-existing-topic">>,
593-
_ = application:stop(wolff), %% ensure stopped
594-
{ok, _} = application:ensure_all_started(wolff),
581+
ok = start_app(),
595582
?assertMatch({error, _}, wolff:ensure_supervised_producers(<<"never-started">>, Topic, #{name => ?FUNCTION_NAME})),
596583
?assertMatch([], supervisor:which_children(wolff_producers_sup)),
597584
ok = wolff:stop_and_delete_supervised_client(ClientId),
@@ -601,8 +588,7 @@ fail_retry_success_test() ->
601588
CntrEventsTable = ets:new(cntr_events, [public]),
602589
wolff_tests:install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),
603590
ClientId = <<"supervised-producers">>,
604-
_ = application:stop(wolff), %% ensure stopped
605-
{ok, _} = application:ensure_all_started(wolff),
591+
ok = start_app(),
606592
ClientCfg = client_config(),
607593
{ok, _ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
608594
ProducerCfg0 = producer_config(?FUNCTION_NAME),
@@ -637,7 +623,7 @@ fail_retry_success_test() ->
637623
after
638624
ok = wolff:stop_and_delete_supervised_producers(Producers),
639625
ok = wolff:stop_and_delete_supervised_client(ClientId),
640-
ok = application:stop(wolff),
626+
ok = stop_app(),
641627
telemetry:detach(tap),
642628
wolff_tests:deinstall_event_logging(?FUNCTION_NAME),
643629
ets:delete(CntrEventsTable),
@@ -651,8 +637,7 @@ fail_retry_failed_test() ->
651637
CntrEventsTable = ets:new(cntr_events, [public]),
652638
wolff_tests:install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),
653639
ClientId = <<"supervised-producers">>,
654-
_ = application:stop(wolff), %% ensure stopped
655-
{ok, _} = application:ensure_all_started(wolff),
640+
ok = start_app(),
656641
ClientCfg = client_config(),
657642
{ok, _ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
658643
ProducerCfg0 = producer_config(?FUNCTION_NAME),
@@ -691,7 +676,7 @@ fail_retry_failed_test() ->
691676
after
692677
ok = wolff:stop_and_delete_supervised_producers(Producers),
693678
ok = wolff:stop_and_delete_supervised_client(ClientId),
694-
ok = application:stop(wolff),
679+
ok = stop_app(),
695680
telemetry:detach(tap),
696681
wolff_tests:deinstall_event_logging(?FUNCTION_NAME),
697682
ets:delete(CntrEventsTable),
@@ -824,3 +809,12 @@ create_topic(Topic, Partitions) ->
824809

825810
delete_topic(Topic) ->
826811
wolff_test_utils:delete_topic(Topic).
812+
813+
stop_app() ->
814+
_ = application:stop(wolff), %% ensure stopped
815+
ok.
816+
817+
start_app() ->
818+
ok = stop_app(),
819+
{ok, _} = application:ensure_all_started(wolff),
820+
ok.

0 commit comments

Comments
 (0)