fix: do not linger unless writing to disk

Author: zmstone

changelog.md

@@ -1,3 +1,9 @@
+* 4.1.1
+  - Do not linger before enqueue when queue is not on disk.
+    Since 4.0.0, the linger is moved from after the queue to before the queue to optimize IOPS in disk mode.
+    This however added unnecessary delay for memory (or offload mode before disk).
+    Now linger happens only when wirting to disk.
+
 * 4.1.0
   - Fix 'failed' telemetry counter double-increment due to race condition. [#102](https://github.com/kafka4beam/wolff/pull/102)
   - Added producer process label `{wolff_producer, KafkaClientId, Topic, Partition}` [#102](https://github.com/kafka4beam/wolff/pull/102)

rebar.config

@@ -1,5 +1,5 @@
 {deps, [ {kafka_protocol, "4.3.0"}
-       , {replayq, "0.4.1"}
+       , {replayq, "0.5.0"}
        , {lc, "0.3.5"}
        , {telemetry, "1.1.0"}
        ]}.

src/wolff_producer.erl

@@ -920,13 +920,20 @@ ensure_linger_expire_timer_cancel(#{?linger_expire_timer := LTimer} = St) ->
   St#{?linger_expire_timer => false}.
 
 %% check if the call collection should continue to linger before enqueue
-is_linger_continue(#{calls := Calls, config := Config}) ->
-  #{max_linger_ms := MaxLingerMs, max_linger_bytes := MaxLingerBytes} = Config,
-  #{ts := Ts, bytes := Bytes} = Calls,
-  case Bytes < MaxLingerBytes of
+is_linger_continue(#{config := #{max_linger_ms := 0}}) ->
+  false;
+is_linger_continue(#{calls := Calls, config := Config, replayq := Q}) ->
+  case replayq:is_writing_to_disk(Q) of
     true ->
-      TimeLeft = MaxLingerMs - (now_ts() - Ts),
-      (TimeLeft > 0) andalso {true, TimeLeft};
+      #{max_linger_ms := MaxLingerMs, max_linger_bytes := MaxLingerBytes} = Config,
+      #{ts := Ts, bytes := Bytes} = Calls,
+      case Bytes < MaxLingerBytes of
+        true ->
+          TimeLeft = MaxLingerMs - (now_ts() - Ts),
+          (TimeLeft > 0) andalso {true, TimeLeft};
+        false ->
+          false
+      end;
     false ->
       false
   end.

test/wolff_tests.erl

@@ -512,7 +512,8 @@ replayq_overflow_test() ->
                   replayq_max_total_bytes => BatchSize,
                   required_acks => all_isr,
                   max_linger_ms => LingerMs, %% delay enqueue
-                  max_linger_bytes => BatchSize + 1 %% delay enqueue
+                  max_linger_bytes => BatchSize + 1, %% delay enqueue
+                  replayq_dir => "test-data/overlow"
                  },
   {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
   Pid = wolff_producers:lookup_producer(Producers, 0),
@@ -577,7 +578,6 @@ replayq_highmem_overflow_test() ->
   ProducerCfg = #{max_batch_bytes => 1, %% make sure not collecting calls into one batch
                   replayq_max_total_bytes => BatchSize*1000,
                   required_acks => all_isr,
-                  max_linger_ms => 1000, %% do not send to kafka immediately
                   drop_if_highmem => true
                  },
   {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
@@ -664,7 +664,7 @@ mem_only_replayq_test() ->
 recover_from_replayq_test() ->
   ClientCfg = client_config(),
   {ok, Client} = start_client(<<"client-2">>, ?HOSTS, ClientCfg),
-  ProducerCfg = #{replayq_dir => "test-data-2"},
+  ProducerCfg = #{replayq_dir => "test-data/recover"},
   {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
   Msg = #{key => ?KEY, value => <<"value">>},
   {0, BaseOffset} = wolff:send_sync(Producers, [Msg], 3000),
@@ -858,7 +858,8 @@ test_batch_split_then_drop(Topic, MaxMessageBytes) ->
                     %% ensure batching by delay enqueue by 100 seconds
                     max_linger_ms => 100,
                     %% ensure linger is not expired by reaching size
-                    max_linger_bytes => MaxMessageBytes * 100
+                    max_linger_bytes => MaxMessageBytes * 100,
+                    replayq_dir => "test-data/batch-split"
                    },
     {ok, Producers} = wolff:start_producers(Client, TopicBin, ProducerCfg),
     MaxBytesCompensateOverhead = MaxMessageBytes - ?BATCHING_OVERHEAD - 7,