Thanks for driving this. The motivation makes sense, and I verified it against the code: retries defaults to Integer.MAX_VALUE, Sender.canRetry() re-enqueues indefinitely on any RetriableException, and flush() bottoms out in an untimed CountDownLatch.await().
Three things I'd want I wanted to point out here: 1. *Idempotence + sequence numbers*. Idempotence defaults on. Expiring a batch that already holds a sequence leaves a gap, and the next batch fails fatally with *OutOfOrderSequenceException*, malforming the bucket. KIP-91 co-designed delivery-timeout with producer-id/epoch handling for this reason. I think we should specify what happens to writer-id/sequence state on expiry and how ordering is preserved for surviving batches. 2. *In-flight vs. queued expiry*. The motivating case (DiskWriteLockedException, acks=all) is a slow in-flight batch the server may persist; failing it to the user risks an app-level retry and a duplicate. We should state explicitly whether we expire in-flight batches or only queued ones, and the trade-off either way. 3. *TimeoutException is retriable-typed* (extends RetriableException). The expiry path must be terminal and bypass canRetry()/reEnqueue(), and canRetry() needs a hasExpired() short-circuit; otherwise, the batch we're expiring loops forever. NIT: I'd not call this "fully backward-compatible". Behaviorally, a transient outage that's survived by blocking today becomes a write failure (and a Flink job restart) under a 2-minute default. We can do something similar to Kafka, but we should also include a prominent upgrade note. Also, we could add an expired-records counter alongside recordsRetryTotal/recordsSendTotal, since the point is surfacing previously-silent failures. Let me know your thoughts. Best, Giannis
