feat: add max_retry to drop a batch after too many failed retries

Add a new producer option `max_retry` (default `infinity`, i.e. retry forever).

When set to a non-negative integer, a batch that keeps receiving Kafka error
responses (e.g. `not_leader_for_partition`) is dropped once its attempt counter
reaches `max_retry + 1` — the initial send plus `max_retry` retries have all
failed. `max_retry = 0` drops on the first error.

The check is applied in the Kafka-error path of `do_handle_kafka_ack/3`: when
the front in-flight request has been attempted enough times, it is popped and
its callers are acked with reason `max_retry_exceeded` (via
`clear_sent_and_ack_callers/3`) and the `dropped` counter is bumped, instead of
incrementing the attempt counter and retrying. The producer still reconnects
afterwards, since the error means the current connection is stale for the
remaining in-flight requests.

This bounds retries on Kafka error responses only; resends triggered purely by
connection loss are bounded by `max_batch_age`, not `max_retry`.

Author: zmstone

changelog.md

@@ -1,3 +1,13 @@
+* 4.3.0
+  - Add `max_retry` producer option (default `infinity`, i.e. retry forever).
+    When set to a non-negative integer, a batch that keeps getting Kafka error
+    responses (e.g. `not_leader_for_partition`) is dropped once it has been
+    attempted `max_retry + 1` times (the initial send plus `max_retry` retries).
+    `max_retry = 0` drops on the first error. Each dropped message has its ack
+    callback evaluated with reason `max_retry_exceeded` and bumps the `dropped`
+    counter. This bounds retries on Kafka error responses; resends triggered
+    purely by connection loss remain bounded by `max_batch_age`, not `max_retry`.
+
 * 4.2.0
   - Add `max_batch_age` producer option (default `infinity`, i.e. disabled).
     When set to a number of milliseconds, a batch is dropped instead of being

include/wolff.hrl

@@ -7,6 +7,7 @@
 -define(buffer_overflow_discarded, buffer_overflow_discarded).
 -define(partition_lost, partition_lost).
 -define(message_expired, message_expired).
+-define(max_retry_exceeded, max_retry_exceeded).
 
 %% Kafka has default batch size limit 1000012.
 %% Since Kafka 2.4, it has been extended to 1048588.

src/wolff_producer.erl

@@ -50,6 +50,7 @@
                       compression |
                       drop_if_highmem |
                       max_batch_age |
+                      max_retry |
                       telemetry_meta_data |
                       max_partitions.
 
@@ -67,6 +68,7 @@
                        compression => kpro:compress_option(),
                        drop_if_highmem => boolean(),
                        max_batch_age => timeout(),
+                       max_retry => infinity | non_neg_integer(),
                        telemetry_meta_data => map(),
                        max_partitions => pos_integer()
                       }.
@@ -84,6 +86,7 @@
                           compression => kpro:compress_option(),
                           drop_if_highmem => boolean(),
                           max_batch_age => timeout(),
+                          max_retry => infinity | non_neg_integer(),
                           telemetry_meta_data => map(),
                           max_partitions => pos_integer()
                          }.
@@ -170,6 +173,15 @@
 %%    on reconnect). Each dropped message has its ack callback evaluated with
 %%    reason `message_expired' and bumps the `dropped' and `dropped_expired'
 %%    counters. Set to `infinity' to disable (never drop by age).
+%% * `max_retry': default `infinity'. Maximum number of times a batch is retried
+%%    after a Kafka error response (e.g. `not_leader_for_partition') before it is
+%%    dropped. A batch is dropped once its attempt counter reaches
+%%    `max_retry + 1' (i.e. the initial send plus `max_retry' retries have all
+%%    failed). Each dropped message has its ack callback evaluated with reason
+%%    `max_retry_exceeded' and bumps the `dropped' counter. `max_retry = 0' drops
+%%    on the first error; `infinity' (the default) retries forever. Note this
+%%    counts Kafka error responses only; resends triggered purely by connection
+%%    loss are bounded by `max_batch_age', not by `max_retry'.
 -spec start_link(wolff:client_id(), topic(), partition(), pid() | ?conn_down(any()), config_in()) ->
   {ok, pid()} | {error, any()}.
 start_link(ClientId, Topic, Partition, MaybeConnPid, Config) ->
@@ -437,6 +449,7 @@ use_defaults(Config) ->
                         {max_send_ahead, 0},
                         {compression, no_compression},
                         {max_batch_age, infinity},
+                        {max_retry, infinity},
                         {reconnect_delay_ms, 2000}
                        ]).
 
@@ -787,20 +800,45 @@ do_handle_kafka_ack(EC, _BaseOffset, St) when EC =:= ?message_too_large orelse E
   end;
 do_handle_kafka_ack(ErrorCode, _BaseOffset, St) ->
   %% Other errors, such as not_leader_for_partition
-  #{sent_reqs := SentReqs} = St,
-  {value, #{attempts := Attempts, q_items := Items}} = queue:peek(SentReqs),
-  #{topic := Topic,
+  #{sent_reqs := SentReqs,
+    topic := Topic,
     partition := Partition,
     config := Config
    } = St,
+  {value, #{attempts := Attempts, q_items := Items}} = queue:peek(SentReqs),
   NrOfCalls = count_calls(Items),
-  inc_sent_failed(Config, NrOfCalls, Attempts),
-  log_warn(Topic, Partition, "error_in_produce_response",
-           #{error_code => ErrorCode,
-             batch_size => count_msgs(Items),
-             attempts => Attempts}),
-  St1 = increment_attempt(St),
-  erlang:throw({kafka_error, ErrorCode, St1}).
+  MaxRetry = maps:get(max_retry, Config, infinity),
+  case is_max_retry_reached(Attempts, MaxRetry) of
+    true ->
+      %% Give up: drop the front batch instead of retrying it again.
+      log_error(Topic, Partition, "dropped_produce_request_reached_max_retry",
+                #{error_code => ErrorCode,
+                  batch_size => count_msgs(Items),
+                  attempts => Attempts,
+                  max_retry => MaxRetry}),
+      wolff_metrics:dropped_inc(Config, NrOfCalls),
+      %% clear_sent_and_ack_callers/3 pops the front request, bumps the
+      %% failed/retried_failed counter, and acks the callers with the reason.
+      St1 = clear_sent_and_ack_callers(ErrorCode, ?max_retry_exceeded, St),
+      %% Still reconnect: the error (e.g. not_leader_for_partition) means the
+      %% current connection is stale for the remaining in-flight requests.
+      erlang:throw({kafka_error, ErrorCode, St1});
+    false ->
+      inc_sent_failed(Config, NrOfCalls, Attempts),
+      log_warn(Topic, Partition, "error_in_produce_response",
+               #{error_code => ErrorCode,
+                 batch_size => count_msgs(Items),
+                 attempts => Attempts}),
+      St1 = increment_attempt(St),
+      erlang:throw({kafka_error, ErrorCode, St1})
+  end.
+
+%% A batch is dropped once it has been attempted `max_retry + 1' times, i.e. the
+%% initial send plus `max_retry' retries have all failed.
+is_max_retry_reached(_Attempts, infinity) ->
+  false;
+is_max_retry_reached(Attempts, MaxRetry) ->
+  Attempts >= MaxRetry + 1.
 
 %% Since we only handle Kafka ack for the first sent request in the sent queue
 %% we only bump the first itme with attempt counter
@@ -1329,6 +1367,15 @@ is_batch_expired_test_() ->
        ?_assert(is_batch_expired(Sent([Item(50)]), 50, Now))}
     ].
 
+is_max_retry_reached_test_() ->
+    [ {"infinity never reached", ?_assertNot(is_max_retry_reached(1000000, infinity))}
+      %% max_retry = 0 -> drop on the first attempt's failure
+    , {"0: first attempt reached", ?_assert(is_max_retry_reached(1, 0))}
+      %% max_retry = 3 -> allow attempts 1..4, drop when attempts reaches 4
+    , {"3: attempt 3 not reached", ?_assertNot(is_max_retry_reached(3, 3))}
+    , {"3: attempt 4 reached", ?_assert(is_max_retry_reached(4, 3))}
+    ].
+
 maybe_log_discard_test_() ->
     [ {"no-increment", fun() -> maybe_log_discard(undefined, 0, ?DISCARD_OVERFLOW) end}
     , {"fake-last-old",

test/wolff_tests.erl

@@ -414,6 +414,52 @@ drop_expired_queued_batch() ->
   ets:delete(CntrEventsTable),
   deinstall_event_logging(?FUNCTION_NAME).
 
+%% With max_retry = 0 a batch is dropped on the first Kafka error response,
+%% instead of being retried, and the caller is acked with `max_retry_exceeded'.
+drop_batch_on_max_retry_test_() ->
+  {timeout, 30, fun drop_batch_on_max_retry/0}.
+
+drop_batch_on_max_retry() ->
+  CntrEventsTable = ets:new(cntr_events, [public]),
+  install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),
+  ClientCfg = client_config(),
+  {ok, Client} = start_client(<<"client-1">>, ?HOSTS, ClientCfg),
+  %% mem-only replayq; max_retry => 0 drops on the first error
+  ProducerCfg = #{max_retry => 0, max_send_ahead => 0, reconnect_delay_ms => 0},
+  {ok, Producers} = wolff:start_producers(Client, <<"test-topic">>, ProducerCfg),
+  Producer = wolff:get_producer(Producers, 0),
+  TesterPid = self(),
+  %% Swallow the produce request so we can inject a fake error response below.
+  ok = meck:new(kpro, [no_history, no_link, passthrough]),
+  meck:expect(kpro, send, fun(_Conn, _Req) -> TesterPid ! sent_to_kafka, ok end),
+  Ref = make_ref(),
+  AckFun = fun(_Partition, Result) -> TesterPid ! {ack, Ref, Result}, ok end,
+  Msg = #{key => ?KEY, value => <<"value">>},
+  try
+    _ = wolff:send(Producers, [Msg], AckFun),
+    receive sent_to_kafka -> ok after 5000 -> error(timeout_waiting_send) end,
+    %% grab the connection and the in-flight request ref from the producer state
+    #{conn := Conn, sent_reqs := SentReqs} = sys:get_state(Producer),
+    {value, #{req_ref := ReqRef}} = queue:peek(SentReqs),
+    %% inject a Kafka error response for that request
+    ErrBody = #{responses =>
+                  [#{partition_responses =>
+                       [#{error_code => not_leader_for_partition,
+                          base_offset => -1}]}]},
+    Producer ! {msg, Conn, #kpro_rsp{api = produce, ref = ReqRef, msg = ErrBody}},
+    receive
+      {ack, Ref, Result} -> ?assertEqual(max_retry_exceeded, Result)
+    after 5000 -> error(timeout_waiting_ack)
+    end
+  after
+    meck:unload(kpro),
+    ok = wolff:stop_producers(Producers),
+    ok = stop_client(Client)
+  end,
+  ?assertEqual([1], get_telemetry_seq(CntrEventsTable, [wolff, dropped])),
+  ets:delete(CntrEventsTable),
+  deinstall_event_logging(?FUNCTION_NAME).
+
 zero_ack_test() ->
   CntrEventsTable = ets:new(cntr_events, [public]),
   install_event_logging(?FUNCTION_NAME, CntrEventsTable, false),