Broker.Close can be blocked for the full broker throttle duration

Description

A long broker-supplied throttle duration can prevent a producer from closing and recovering after the connection has failed.

We observed a ProduceResponse with a throttle duration of 12h42m52.356s. Sarama honored the response by entering waitIfThrottled. Approximately two seconds later, an in-flight request hit a TCP read timeout and the broker producer changed to the closing state, but the broker could not finish closing until the throttle timer expired. Synchronous producers remained blocked during that period, and production did not recover even after the Kafka cluster was healthy again.

Sanitized logs:

broker/242559 *sarama.ProduceResponse throttled 12h42m52.356s
broker/242559 waiting for throttle timer
producer/broker/242559 state change to [closing] because read tcp <client>-><broker>: i/o timeout

A goroutine dump showed (*syncProducer).SendMessage blocked waiting for its result while the broker producer was unable to complete its close and retry path.

Versions

The same throttle implementation is still present on the current main branch and in v1.60.1.

Root cause

sendInternal calls waitIfThrottled while its caller holds Broker.lock.

waitIfThrottled blocks on <-b.throttleTimer.C for the complete broker-supplied duration. Since PR #2826, it also holds throttleTimerLock for the complete wait.

When an in-flight request fails while another send is waiting for the throttle timer:

1. the broker producer starts its connection error/close path;
2. Broker.Close tries to acquire Broker.lock;
3. the send that owns Broker.lock remains blocked on the throttle timer;
4. close, reconnect, retry, and delivery-result propagation cannot progress until the timer fires.

Net.ReadTimeout, Net.WriteTimeout, and Producer.Timeout do not bound this wait because no network operation is active in waitIfThrottled.

Expected behavior

Sarama should continue to honor a broker throttle while the broker connection is active.

When the broker is closing or the connection has failed, the local throttle wait should be interrupted promptly. The blocked send should return an existing connection/closed error so the normal producer retry, metadata refresh, and reconnect paths can proceed.

Actual behavior

The broker close and producer recovery paths can remain blocked for the complete broker-supplied throttle duration.

Proposed scope

This issue does not propose ignoring, shortening, or globally capping valid Kafka quota throttles.

A focused fix should:

• make the throttle wait interruptible by broker close or connection failure;
• avoid holding the throttle mutex while blocked on the timer;
• ensure the send path does not prevent Broker.Close from making progress;
• preserve the existing throttle behavior while the connection remains active;
• add regression coverage for a long throttle combined with concurrent close/connection failure;
• pass go test -race without leaking goroutines.

Related issues and pull requests

• #2823 reported the race between setThrottle and waitIfThrottled.
• #2826 fixed that race by adding throttleTimerLock, but did not make the wait interruptible.
• #2121 reported producer shutdown remaining blocked for a long time.
• #2133 fixed a different Broker.Close deadlock in the broker producer error path and is useful precedent for keeping close paths non-blocking.