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

[zmstone]

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

Motivation

Today wolff retries a batch indefinitely on Kafka error responses (e.g. not_leader_for_partition). There is no way to cap the number of retries and give up on a batch. max_batch_age (added in 4.2.0) bounds time, but some callers want to bound the retry count on error responses.

What this does

When max_retry is set to a non-negative integer, a batch is dropped once its attempt counter reaches max_retry + 1 — i.e. the initial send plus max_retry retries have all failed:

• max_retry = 0 → drop on the first error response.
• max_retry = 3 → drop after the 4th failed attempt (3 retries).
• infinity (default) → retry forever (unchanged behavior).

The check lives 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 the existing clear_sent_and_ack_callers/3, which also bumps failed/retried_failed), and the dropped counter is bumped — instead of incrementing the attempt counter and retrying. The producer still reconnects afterwards, since the error (e.g. not_leader_for_partition) means the current connection is stale for the remaining in-flight requests.

Scope / notes

• This bounds retries on Kafka error responses only. Resends triggered purely by connection loss do not increment the attempts counter and are bounded by max_batch_age, not max_retry. The two options are complementary (time-based vs error-count-based).
• New public reason atom max_retry_exceeded reaches the ack callback; downstream callers that pattern-match ack results should add a clause for it.

Tests

• is_max_retry_reached_test_/0 — pure unit tests of the boundary (infinity, 0, and the max_retry + 1 edge).
• drop_batch_on_max_retry_test_/0 — with max_retry = 0, injects a not_leader_for_partition produce response and asserts the batch is dropped (ack max_retry_exceeded, dropped bumped) rather than retried.

Ran rebar3 eunit locally: the new tests pass; the only failures are 4 pre-existing environment-specific tests (message_too_large, record_list_too_large, one_byte_limit, leader_restart) that also fail on unmodified main in this environment and pass in CI.