Skip to content

fix: do not linger unless writing to disk #106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zmstone merged 1 commit into from
Nov 3, 2025
Merged

fix: do not linger unless writing to disk #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions changelog.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
Expand Down
2 changes: 1 addition & 1 deletion rebar.config
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"}
]}.
Expand Down
19 changes: 13 additions & 6 deletions src/wolff_producer.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}) ->

@kiliangrashoff kiliangrashoff Nov 7, 2025 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@zmstone thanks for implementing this, I think it will help us to get faster PUBACKs when using the EMQX Cloud Kafka integration with hybrid buffer mode and max_linger_ms set to a non-0 value.

I see that is_writing_to_disk will return true when the tail segments are written to disk in offload mode. I am afraid that this will cause PUBACK latency to increase due to linger before append when Kafka is unavailable. Would it be possible to also skip linger before append when replayq offloads to disk? I think should not be necessary to linger when offload mode writes to disk: append should still write to memory, but tail elements are written to disk, which should allow batching disk writes.

An additional question: we need our messages to Kafka to be batched to avoid increasing load on our Kafka cluster. Will all messages that are queued during a timeframe of max_linger_ms still be batched into a single message at the popping end of the queue, or does this change cause the producer to send messages to Kafka more frequently? The desired behavior for us is that in hybrid mode the producer still lingers, but after enqueueing the messages, decoupling PUBACK latency from kafka message batching.

@zmstone zmstone Nov 7, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @kiliangrashoff

For EMQX, PUBACK latency increase to cause back-pressure towards MQTT client, otherwise the memory part will eventually overflow.

For Kafka load:
There are other ways to throttle requests towards Kafka.
The most effective is to lower the send-ahead counter.
EMQX has this config named "Max Inflight" from the UI and max_inflight in config file.
The default value is 10, if you lower to 1, there is only going to be 1 request sent to Kafka before Kafka acknowledges. Hence the pending ones will effectively linger (and form a larger batch).

The unconditional linger will add unnecessary latency even when Kafka and Kafka clients are both idling.

It's like the 40m delay of TCP stack without NO_DELAY flag.

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.
Expand Down
9 changes: 5 additions & 4 deletions test/wolff_tests.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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,
Expand Down