Merge remote-tracking branch 'origin/main-4.1' into 260723-sync-main-4.1

# Conflicts:
#	changelog.md
#	src/wolff_producer.erl

Author: zmstone

changelog.md

@@ -19,6 +19,20 @@
     buffered messages across long broker outages instead of sending arbitrarily
     stale data.
 
+* 4.1.11
+  - `max_linger_ms` now also applies to memory-mode (and not-yet-offloaded) queues.
+    Since 4.1.1, the linger delay was only applied before enqueue, and only when
+    replayq is writing to disk (to batch disk writes), meaning memory-mode producers
+    sent each arrival immediately, resulting in many small produce requests under
+    steady low-to-moderate load.
+    Now, when it is time to form a new produce request and the queue holds less than
+    `min(max_linger_bytes, max_batch_bytes)` bytes, the producer waits up to
+    `max_linger_ms` for more calls to accumulate before popping the queue.
+    The wait ends immediately once a full batch's worth of bytes is queued, so under
+    sustained load full batches are sent with no added latency; only under-sized
+    (tail) batches are delayed, bounded by `max_linger_ms`.
+    The default `max_linger_ms = 0` keeps the previous behavior (send immediately).
+
 * 4.1.10
   - Reserve a minimum producer buffer under high memory pressure.
     When `drop_if_highmem` is enabled and the system reports high memory usage,

src/wolff_producer.erl

@@ -95,6 +95,8 @@
 -define(reconnect, reconnect).
 -define(linger_expire, linger_expire).
 -define(linger_expire_timer, linger_expire_timer).
+-define(pop_linger_expire, pop_linger_expire).
+-define(pop_linger_timer, pop_linger_timer).
 -define(DEFAULT_REPLAYQ_SEG_BYTES, 10 * 1024 * 1024).
 -define(DEFAULT_REPLAYQ_LIMIT, 2000000000).
 -define(Q_ITEM(CallId, Ts, Batch), {CallId, Ts, Batch}).
@@ -137,6 +139,7 @@
                   , topic := topic()
                   , calls := ?EMPTY | calls()
                   , sender => pid()
+                  , ?pop_linger_timer := false | reference()
                   }.
 
 %% @doc Start a per-partition producer worker.
@@ -150,8 +153,12 @@
 %% * `max_batch_bytes': Most number of bytes to collect into a produce request.
 %%    NOTE: This is only a rough estimation of the batch size towards kafka, NOT the
 %%    exact max allowed message size configured in kafka.
-%% * `max_linger_ms': Age in milliseconds a baatch can stay in queue when the connection
-%%    is idle (as in no pending acks). Default: 0 (as in send immediately)
+%% * `max_linger_ms': Number of milliseconds to wait for more calls to accumulate
+%%    a larger batch. When the queue is writing to disk, the wait happens before
+%%    enqueue (to also batch disk writes); otherwise the producer delays popping
+%%    the queue for a new produce request until at least
+%%    `min(max_linger_bytes, max_batch_bytes)' bytes are queued, or the linger
+%%    time expires. Default: 0 (as in send immediately)
 %% * `max_linger_bytes': Number of bytes to collect before sending it to Kafka.
 %%    If set to 0, `max_batch_bytes' is taken for mem-only mode, otherwise it's 10 times
 %%    `max_batch_bytes' (but never exceeds 10MB) to optimize disk write.
@@ -190,7 +197,8 @@ start_link(ClientId, Topic, Partition, MaybeConnPid, Config) ->
          partition => Partition,
          conn => MaybeConnPid,
          config => use_defaults(Config),
-         ?linger_expire_timer => false
+         ?linger_expire_timer => false,
+         ?pop_linger_timer => false
         },
   %% the garbage collection can be expensive if using the default 'on_heap' option.
   SpawnOpts = [{spawn_opt, [{message_queue_data, off_heap}]}],
@@ -334,6 +342,16 @@ handle_info(?linger_expire, St0) ->
   St1 = enqueue_calls(St0#{?linger_expire_timer => false}, no_linger),
   St = maybe_send_to_kafka(St1),
   {noreply, St};
+handle_info({timeout, Ref, ?pop_linger_expire}, St0) ->
+  case maps:get(?pop_linger_timer, St0, false) of
+    Ref ->
+      %% Lingered long enough, flush the queue even if the batch is under-sized
+      St = maybe_send_to_kafka(St0#{?pop_linger_timer => false}, no_linger),
+      {noreply, St};
+    _ ->
+      %% stale timer message: the timer fired right before it was cancelled
+      {noreply, St0}
+  end;
 handle_info(?SEND_REQ(_, Batch, _) = Call, #{calls := Calls0, config := #{max_linger_bytes := Max}} = St0) ->
   Bytes = batch_bytes(Batch),
   Calls = collect_send_calls(Call, Bytes, Calls0, Max),
@@ -400,7 +418,9 @@ handle_leader_connection(#{topic := Topic, partition := Partition, conn := Conn}
              conn := Conn},
   St2 = get_produce_version(St1),
   St3 = resend_sent_reqs(St2, ?reconnect),
-  maybe_send_to_kafka(St3);
+  %% no_linger: the queue had the whole disconnected period to accumulate,
+  %% flush it immediately (also in case the pop-linger expired while disconnected)
+  maybe_send_to_kafka(St3, no_linger);
 handle_leader_connection(#{conn := Reason} = St) ->
   mark_connection_down(St#{reconnect_timer => ?no_timer}, Reason).
 
@@ -619,15 +639,22 @@ discard_expired_front(#{replayq := Q,
   ok = maybe_log_discard(St, NrOfCalls, ?DISCARD_EXPIRED),
   St#{replayq := NewQ, pending_acks := NewPendingAcks}.
 
-maybe_send_to_kafka(St0) ->
+maybe_send_to_kafka(St) ->
+  maybe_send_to_kafka(St, maybe_linger).
+
+%% Linger =:= no_linger when the pop-linger has expired or after a reconnect:
+%% flush the queue without waiting for more calls to accumulate.
+maybe_send_to_kafka(St0, Linger) ->
+  %% shed expired batches first, so the pop-linger gate below
+  %% sees the queue size without them
   #{conn := Conn, replayq := Q} = St = drop_expired_head(St0),
   case replayq:count(Q) =:= 0 of
     true ->
       %% nothing to send
       St;
     false when is_pid(Conn) ->
       %% has connection, try send
-      maybe_send_to_kafka_has_pending(St);
+      maybe_send_to_kafka_has_pending(St, Linger);
     false ->
       %% no connection
       %% maybe the connection was closed after ideling
@@ -636,9 +663,13 @@ maybe_send_to_kafka(St0) ->
       ensure_delayed_reconnect(St, no_delay_for_first_attempt)
   end.
 
-maybe_send_to_kafka_has_pending(St) ->
+maybe_send_to_kafka_has_pending(St, Linger) ->
   case is_send_ahead_allowed(St) of
-    true -> send_to_kafka(St);
+    true ->
+      case should_linger_before_pop(St, Linger) of
+        true -> ensure_pop_linger_timer(St);
+        false -> send_to_kafka(ensure_pop_linger_cancel(St))
+      end;
     false -> St %% reached max inflight limit
   end.
 
@@ -1106,6 +1137,44 @@ is_linger_continue(#{calls := Calls, config := Config, replayq := Q}) ->
       false
   end.
 
+%% Check if the producer should delay popping the queue so an under-sized
+%% batch gets a chance to grow from future send calls.
+%% Only when not writing to disk: when writing to disk, the linger is applied
+%% before enqueue (see is_linger_continue/1) to also batch disk writes.
+%% NOTE: the ?pop_linger_timer key may be absent when the new beam is
+%% hot-loaded into an already running producer, hence maps:get/3 with
+%% default in ensure_pop_linger_* below.
+should_linger_before_pop(_St, no_linger) ->
+  false;
+should_linger_before_pop(#{config := #{max_linger_ms := 0}}, _) ->
+  false;
+should_linger_before_pop(#{replayq := Q, config := Config}, _) ->
+  #{max_linger_bytes := MaxLingerBytes, max_batch_bytes := MaxBatchBytes} = Config,
+  (not replayq:is_writing_to_disk(Q)) andalso
+    replayq:bytes(Q) < min(MaxLingerBytes, MaxBatchBytes).
+
+ensure_pop_linger_timer(St) ->
+  case maps:get(?pop_linger_timer, St, false) of
+    false ->
+      #{config := #{max_linger_ms := Ms}} = St,
+      Ref = erlang:start_timer(Ms, self(), ?pop_linger_expire),
+      St#{?pop_linger_timer => Ref};
+    _ ->
+      %% timer is already started
+      St
+  end.
+
+ensure_pop_linger_cancel(St) ->
+  case maps:get(?pop_linger_timer, St, false) of
+    false ->
+      St;
+    Ref ->
+      %% no need to flush a stale ?pop_linger_expire message:
+      %% it's ignored by the timer reference check in handle_info
+      _ = erlang:cancel_timer(Ref),
+      St#{?pop_linger_timer => false}
+  end.
+
 enqueue_calls(#{calls := ?EMPTY} = St, _) ->
   %% no call to enqueue
   St;

test/wolff_tests.erl

@@ -19,6 +19,9 @@
 
 -define(KEY, key(?FUNCTION_NAME)).
 -define(HOSTS, [{"localhost", 9092}]).
+%% Allowed measurement slack when asserting that at least
+%% max_linger_ms has elapsed (timer precision, scheduling delays).
+-define(LINGER_SLACK_MS, 100).
 -define(assert_eq_optional_tail(EXPR,EXPECTED), assert_eq_optional_tail(fun() -> EXPR end, EXPECTED)).
 
 %% Kafka v2 batch consists of below fields:
@@ -791,6 +794,189 @@ replayq_highmem_overflow_test() ->
   ets:delete(CntrEventsTable),
   deinstall_event_logging(?FUNCTION_NAME).
 
+%% In memory mode, the producer should linger before popping the queue:
+%% messages arriving one-by-one (each handled in its own mailbox message)
+%% are collected into a single produce request which is
+%% sent when max_linger_ms expires since the first message.
+pop_linger_batch_test_() ->
+  {timeout, 30, fun pop_linger_batch/0}.
+
+pop_linger_batch() ->
+  ClientCfg = client_config(),
+  {ok, Client} = start_client(<<"client-1">>, ?HOSTS, ClientCfg),
+  LingerMs = 500,
+  ProducerCfg = #{partitioner => fun(_, _) -> 0 end,
+                  required_acks => all_isr,
+                  max_linger_ms => LingerMs
+                 },
+  {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
+  TesterPid = self(),
+  ok = meck:new(kpro, [no_history, no_link, passthrough]),
+  meck:expect(kpro, send,
+              fun(Conn, Req) ->
+                      Payload = iolist_to_binary(lists:last(tuple_to_list(Req))),
+                      TesterPid ! {sent_to_kafka, Payload},
+                      meck:passthrough([Conn, Req])
+              end),
+  Values = [<<"pop-linger-a">>, <<"pop-linger-b">>, <<"pop-linger-c">>],
+  AckFun = fun(_Partition, _BaseOffset) -> TesterPid ! ack, ok end,
+  T0 = erlang:monotonic_time(millisecond),
+  lists:foreach(
+    fun(V) ->
+        _ = wolff:send(Producers, [#{key => <<>>, value => V}], AckFun),
+        timer:sleep(50) %% ensure calls are not collected from mailbox at once
+    end, Values),
+  try
+    %% expect one single produce request holding all 3 values,
+    %% sent only after the linger time expired
+    receive
+      {sent_to_kafka, Payload} ->
+        Elapsed = erlang:monotonic_time(millisecond) - T0,
+        ?assert(Elapsed >= LingerMs - ?LINGER_SLACK_MS, #{elapsed => Elapsed}),
+        lists:foreach(
+          fun(V) -> ?assertNotEqual(nomatch, binary:match(Payload, V)) end,
+          Values)
+    after
+      3000 -> error(timeout)
+    end,
+    %% all 3 calls are acked from the single produce request
+    ok = wait_for_acks(length(Values)),
+    %% and there is no other produce request
+    receive
+      {sent_to_kafka, _} = Extra -> error({unexpected, Extra})
+    after
+      200 -> ok
+    end,
+    %% linger timer is cleared after the pop
+    Pid = wolff_producers:lookup_producer(Producers, 0),
+    ?assertMatch(#{pop_linger_timer := false}, sys:get_state(Pid))
+  after
+    meck:unload(kpro),
+    ok = wolff:stop_producers(Producers),
+    ok = stop_client(Client)
+  end.
+
+%% The pop-linger must not delay sending when the queue already holds
+%% min(max_linger_bytes, max_batch_bytes) worth of bytes.
+pop_linger_full_batch_no_wait_test_() ->
+  {timeout, 30, fun pop_linger_full_batch_no_wait/0}.
+
+pop_linger_full_batch_no_wait() ->
+  ClientCfg = client_config(),
+  {ok, Client} = start_client(<<"client-1">>, ?HOSTS, ClientCfg),
+  LingerMs = 5000,
+  Msg = #{key => <<>>, value => <<"pop-linger-full-batch">>},
+  MsgBytes = wolff_producer:batch_bytes([Msg]),
+  ProducerCfg = #{partitioner => fun(_, _) -> 0 end,
+                  required_acks => all_isr,
+                  max_linger_ms => LingerMs,
+                  %% the 2nd message accumulates enough bytes to flush
+                  max_linger_bytes => MsgBytes + 1
+                 },
+  {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
+  TesterPid = self(),
+  ok = meck:new(kpro, [no_history, no_link, passthrough]),
+  meck:expect(kpro, send,
+              fun(Conn, Req) ->
+                      Payload = iolist_to_binary(lists:last(tuple_to_list(Req))),
+                      TesterPid ! {sent_to_kafka, Payload},
+                      meck:passthrough([Conn, Req])
+              end),
+  AckFun = fun(_Partition, _BaseOffset) -> TesterPid ! ack, ok end,
+  T0 = erlang:monotonic_time(millisecond),
+  _ = wolff:send(Producers, [Msg], AckFun),
+  _ = wolff:send(Producers, [Msg], AckFun),
+  try
+    %% both messages are flushed in one produce request,
+    %% well before the linger time expires
+    receive
+      {sent_to_kafka, Payload} ->
+        Elapsed = erlang:monotonic_time(millisecond) - T0,
+        ?assert(Elapsed < LingerMs div 2, #{elapsed => Elapsed}),
+        ?assertEqual(2, length(binary:matches(Payload, <<"pop-linger-full-batch">>)))
+    after
+      3000 -> error(timeout)
+    end,
+    ok = wait_for_acks(2)
+  after
+    meck:unload(kpro),
+    ok = wolff:stop_producers(Producers),
+    ok = stop_client(Client)
+  end.
+
+%% A lone under-sized message is delivered within max_linger_ms, never stalls.
+pop_linger_lone_message_test_() ->
+  {timeout, 30, fun pop_linger_lone_message/0}.
+
+pop_linger_lone_message() ->
+  ClientCfg = client_config(),
+  {ok, Client} = start_client(<<"client-1">>, ?HOSTS, ClientCfg),
+  LingerMs = 300,
+  ProducerCfg = #{partitioner => fun(_, _) -> 0 end,
+                  required_acks => all_isr,
+                  max_linger_ms => LingerMs
+                 },
+  {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
+  TesterPid = self(),
+  AckFun = fun(_Partition, _BaseOffset) -> TesterPid ! ack, ok end,
+  T0 = erlang:monotonic_time(millisecond),
+  _ = wolff:send(Producers, [#{key => <<>>, value => <<"pop-linger-lone">>}], AckFun),
+  try
+    ok = wait_for_acks(1),
+    Elapsed = erlang:monotonic_time(millisecond) - T0,
+    ?assert(Elapsed >= LingerMs - ?LINGER_SLACK_MS, #{elapsed => Elapsed})
+  after
+    ok = wolff:stop_producers(Producers),
+    ok = stop_client(Client)
+  end.
+
+%% When the pop-linger expires while the connection is down, the message
+%% stays queued, and is flushed upon reconnect without lingering another time.
+pop_linger_expire_while_disconnected_test_() ->
+  {timeout, 30, fun pop_linger_expire_while_disconnected/0}.
+
+pop_linger_expire_while_disconnected() ->
+  ClientCfg = client_config(),
+  {ok, Client} = start_client(<<"client-1">>, ?HOSTS, ClientCfg),
+  LingerMs = 300,
+  ProducerCfg = #{partitioner => fun(_, _) -> 0 end,
+                  required_acks => all_isr,
+                  max_linger_ms => LingerMs,
+                  reconnect_delay_ms => 1000
+                 },
+  {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
+  Pid = wolff_producers:lookup_producer(Producers, 0),
+  #{conn := Conn} = sys:get_state(Pid),
+  ?assert(is_pid(Conn)),
+  TesterPid = self(),
+  AckFun = fun(_Partition, _BaseOffset) -> TesterPid ! ack, ok end,
+  %% an under-sized message starts the pop-linger timer
+  _ = wolff:send(Producers, [#{key => <<>>, value => <<"pop-linger-disc">>}], AckFun),
+  exit(Conn, kill),
+  %% wait for the linger to expire while disconnected
+  timer:sleep(LingerMs * 2),
+  try
+    %% linger expired, but still disconnected: the message is still queued
+    #{conn := Conn2, replayq := Q} = sys:get_state(Pid),
+    ?assertNot(is_pid(Conn2)),
+    ?assertEqual(1, replayq:count(Q)),
+    %% flushed upon reconnect
+    ok = wait_for_acks(1),
+    ?assertMatch(#{pop_linger_timer := false}, sys:get_state(Pid))
+  after
+    ok = wolff:stop_producers(Producers),
+    ok = stop_client(Client)
+  end.
+
+wait_for_acks(0) ->
+  ok;
+wait_for_acks(N) ->
+  receive
+    ack -> wait_for_acks(N - 1)
+  after
+    5000 -> error(timeout)
+  end.
+
 mem_only_replayq_test() ->
   CntrEventsTable = ets:new(cntr_events, [public]),
   install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),