Sync 1.5.11 to 1.7.7

changelog.md

@@ -1,3 +1,6 @@
+* 1.7.7 (merged 1.5.11)
+  - Fixed a try catch pattern in `gen_server` call towards client process, this should prevent `wolff_producers` from crash if `wolff_client` is killed during initialization. [#49](https://github.com/kafka4beam/wolff/pull/49)
+  - Enhance: use `off_heap` spawn option in producer processes for better gc performance. [#47](https://github.com/kafka4beam/wolff/pull/47)
 * 1.7.6
   - Expose wolff:check_if_topic_exists/3 for checking if a topic is created. [#45](https://github.com/kafka4beam/wolff/pull/45)
 * 1.7.5

src/wolff.app.src

@@ -1,6 +1,6 @@
 {application, wolff,
  [{description, "Kafka's publisher"},
-  {vsn, "1.7.6"},
+  {vsn, "1.7.7"},
   {registered, []},
   {applications,
    [kernel,

src/wolff.appup.src

@@ -1,5 +1,5 @@
 %% -*- mode: erlang; -*-
-{"1.7.6",
+{"1.7.7",
   [
   ],
   [

src/wolff_client.erl

@@ -128,7 +128,7 @@ check_if_topic_exists(Pid, Topic) when is_pid(Pid) ->
 
 safe_call(Pid, Call) ->
   try gen_server:call(Pid, Call, infinity)
-  catch error : Reason -> {error, Reason}
+  catch exit : Reason -> {error, Reason}
   end.
 
 %% request client to send Pid the leader connection.

src/wolff_producer.erl

@@ -117,7 +117,9 @@ start_link(ClientId, Topic, Partition, MaybeConnPid, Config) ->
          config => use_defaults(Config),
          ?linger_expire_timer => false
         },
-  gen_server:start_link(?MODULE, St, []).
+  %% the garbage collection can be expensive if using the default 'on_heap' option.
+  SpawnOpts = [{spawn_opt, [{message_queue_data, off_heap}]}],
+  gen_server:start_link(?MODULE, St, SpawnOpts).
 
 stop(Pid) ->
   gen_server:call(Pid, stop, infinity).

src/wolff_producers.erl

@@ -215,7 +215,8 @@ handle_info(?refresh_partition_count, #{refresh_tref := Tref, config := Config}
     St = refresh_partition_count(St0),
     {noreply, St#{refresh_tref := start_partition_refresh_timer(Config)}};
 handle_info(?rediscover_client, #{client_id := ClientId,
-                                  client_pid := false
+                                  client_pid := false,
+                                  topic := Topic
                                  } = St0) ->
   St1 = St0#{?rediscover_client_tref => false},
   case wolff_client_sup:find_client(ClientId) of
@@ -227,16 +228,18 @@ handle_info(?rediscover_client, #{client_id := ClientId,
       {noreply, St};
     {error, Reason} ->
       log_error("failed_to_discover_client",
-                #{reason => Reason, client_id => ClientId}),
+                #{reason => Reason, topic => Topic, client_id => ClientId}),
       {noreply, ensure_rediscover_client_timer(St1)}
   end;
 handle_info(?init_producers, St) ->
   %% this is a retry of last failure when initializing producer procs
   {noreply, maybe_init_producers(St)};
 handle_info({'DOWN', _, process, Pid, Reason}, #{client_id := ClientId,
-                                                 client_pid := Pid
+                                                 client_pid := Pid,
+                                                 topic := Topic
                                                 } = St) ->
   log_error("client_pid_down", #{client_id => ClientId,
+                                 topic => Topic,
                                  client_pid => Pid,
                                  reason => Reason}),
   %% client down, try to discover it after a delay

test/wolff_tests.erl

@@ -567,6 +567,41 @@ test_leader_restart() ->
   end,
   ok.
 
+%% wolff_client died by the time when wolff_producers tries to initialize producers
+%% it should not cause the wolff_producers_sup to restart
+producers_init_failure_test_() ->
+  {timeout, 10, fun() -> test_producers_init_failure() end}.
+
+test_producers_init_failure() ->
+   _ = application:stop(wolff),
+  {ok, _} = application:ensure_all_started(wolff),
+  ClientId = <<"proucers-init-failure-test">>,
+  Topic = <<"test-topic">>,
+  ClientCfg = #{connection_strategy => per_broker},
+  {ok, ClientPid} = wolff:ensure_supervised_client(ClientId, ?HOSTS, ClientCfg),
+  %% suspend the client so it will not respond to calls
+  sys:suspend(ClientPid),
+  ProducerCfg = #{required_acks => all_isr,
+                  partition_count_refresh_interval_seconds => 0
+                 },
+  %% this call will hang until the client pid is killed later
+  {TmpPid, _} = spawn_monitor(fun() -> {error, {killed, _}} = wolff_producers:start_linked_producers(ClientId, Topic, ProducerCfg), exit(normal) end),
+  %% wait a bit to ensure the spanwed process gets the chance to run
+  timer:sleep(100),
+  %% kill the client pid, so the gen_server:call towards the client will fail
+  exit(ClientPid, kill),
+  receive
+    {'DOWN', _, process, TmpPid, Reason} ->
+      ?assertEqual(normal, Reason)
+  after 2000 ->
+    error(timeout)
+  end,
+  %% cleanup
+  ok = wolff:stop_and_delete_supervised_client(ClientId),
+  ?assertEqual([], supervisor:which_children(wolff_client_sup)),
+  ok = application:stop(wolff),
+  ok.
+
 %% helpers
 
 %% verify wolff_client state upgrade.