fix: flaky test

src/wolff_client.erl

@@ -427,6 +427,7 @@ tr_reason({{Host, Port}, Reason}) ->
     {bin([bin(Host), ":", bin(Port)]), do_tr_reason(Reason)}.
 
 do_tr_reason({timeout, _Stack}) -> connection_timed_out;
+do_tr_reason({{_KproReq, timeout}, _Stack}) -> connection_timed_out;
 do_tr_reason({enetunreach, _Stack}) -> unreachable_host;
 do_tr_reason({econnrefused, _Stack}) -> connection_refused;
 do_tr_reason({nxdomain, _Stack}) -> unresolvable_hostname;

test/wolff_supervised_tests.erl

@@ -6,7 +6,10 @@
 -define(KEY, key(?FUNCTION_NAME)).
 -define(HOSTS, [{"localhost", 9092}]).
 
-supervised_client_test() ->
+supervised_client_test_() ->
+  {timeout, 60, fun() -> do_supervised_client_test() end}.
+
+do_supervised_client_test() ->
   ClientId = <<"supervised-wolff-client">>,
   _ = application:stop(wolff), %% ensure stopped
   {ok, _} = application:ensure_all_started(wolff),
@@ -18,6 +21,8 @@ supervised_client_test() ->
   ProducerCfg0 = producer_config(),
   ProducerCfg = ProducerCfg0#{required_acks => leader_only},
   {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
+  Producer = wolff:get_producer(Producers, 0),
+  _ = wait_for_connection(Producer, 30000),
   Msg = #{key => ?KEY, value => <<"value">>},
   {Partition, BaseOffset} = wolff:send_sync(Producers, [Msg], 3000),
   io:format(user, "\nmessage produced to partition ~p at offset ~p\n",
@@ -170,7 +175,7 @@ stop_with_name_test() ->
   ok.
 
 partition_count_refresh_test_() ->
-  {timeout, 30, %% it takes time to alter topic via cli in docker container
+  {timeout, 120, %% it takes time to alter topic via cli in docker container
    fun test_partition_count_refresh/0}.
 
 test_partition_count_refresh() ->
@@ -229,6 +234,23 @@ start_producers_with_dead_client_test() ->
   ok.
 
 %% helpers
+wait_for_connection(Producer, Time) when Time > 0 ->
+  case get_proc_state(Producer) of
+    #{conn := Conn} when is_pid(Conn) ->
+      Conn;
+    _ ->
+      timer:sleep(50),
+      wait_for_connection(Producer, Time - 50)
+  end;
+wait_for_connection(_Producer, _Time) ->
+  erlang:error({timeout, "wait for connection"}).
+
+get_proc_state(Pid) ->
+  case sys:get_state(Pid) of
+    #{batch_callback_state := St} -> St;
+    St -> St
+  end.
+
 wait_for_pid(F) ->
   Pid = F(),
   case is_pid(Pid) of

test/wolff_tests.erl

@@ -461,7 +461,7 @@ fail_to_connect_all_test() ->
   ok = application:stop(wolff).
 
 leader_restart_test_() ->
-  {timeout, 60,
+  {timeout, 120, %% need a large timeout as running this on PC would take much longer than on CI
    fun() -> test_leader_restart() end}.
 
 test_leader_restart() ->
@@ -472,6 +472,7 @@ test_leader_restart() ->
   %% replication factor has to be 1 to trigger leader_not_available error code
   ReplicationFactor = 1,
   ok = create_topic(Topic, Partitions, ReplicationFactor),
+  timer:sleep(1000), %% sometimes it takes a while for topic to be really ready
   try
     _ = application:stop(wolff), %% ensure stopped
     {ok, _} = application:ensure_all_started(wolff),