Sender: KeyError from muted.remove() while expiring batches orphans the rest; flush() hangs forever

Start in kafka/producer/sender.py at _send_producer_data and _dispatch_error, then inspect RecordAccumulator's muted handling and abort_batches for the existing remove-if-present pattern. Run the reproduction from the issue and add a regression test covering multiple expired batches with guarantee_message_order enabled. Done means every expired batch completes its future and flush() returns without the KeyError or an orphaned batch.

Summary

When brokers are unreachable long enough for batches to reach delivery_timeout_ms, Sender._send_producer_data expires them in a loop. For each one, _dispatch_error ends with:

if self.config['guarantee_message_order']:
    self._accumulator.muted.remove(batch.topic_partition)

A batch that expires while still in the accumulator was never drained, so its partition is not in muted and set.remove raises KeyError. The exception escapes the for expired_batch in expired_batches loop. The remaining batches in that list have already been popped from the accumulator deques / in-flight heap, so they are never completed, retried, or expired again. Their futures never resolve, and KafkaProducer.flush() (no timeout by default) blocks forever.

Sender.run() catches the exception and logs Uncaught error in kafka producer I/O thread, so the I/O thread keeps running and the producer looks healthy.

guarantee_message_order is forced to True when enable_idempotence=True, which is the 3.x default, so default-configured producers are affected.

Versions

• kafka-python 3.0.10 and 3.0.11 (production), and master at cd3f7938e1 (3.1.0.dev) with the reproduction below
• Python 3.10 (production), 3.13 (reproduction)
• Brokers: Kafka 3.1

Reproduction

No broker needed. Uses a real short delivery_timeout_ms because _can_retry() checks the delivery timeout against time.monotonic().

"""Repro: with guarantee_message_order=True, expiring >1 accumulator batch in one pass
raises KeyError from muted.remove() and orphans the remaining expired batches forever."""
import time
from unittest import mock

import kafka
import kafka.errors as Errors
from kafka.cluster import ClusterMetadata
from kafka.producer.record_accumulator import RecordAccumulator
from kafka.producer.sender import Sender
from kafka.structs import TopicPartition


def run(muted_set=None):
    accumulator = RecordAccumulator(delivery_timeout_ms=100, request_timeout_ms=50, linger_ms=0)
    if muted_set is not None:
        accumulator.muted = muted_set
    # guarantee_message_order=True is what enable_idempotence=True (the 3.x default) sets.
    sender = Sender(mock.MagicMock(), ClusterMetadata(), accumulator, guarantee_message_order=True)

    futures = [accumulator.append(TopicPartition('foo', p), 0, b'k', b'v', [])[0] for p in (0, 1)]

    # No leader is known (broker unreachable), so nothing is drained and both batches
    # expire in the same pass. Real clock: _can_retry() checks delivery timeout via time.monotonic().
    time.sleep(0.2)
    for _ in range(3):  # later passes never revisit the orphaned batch
        try:
            sender._send_producer_data()
        except KeyError as e:
            print('  _send_producer_data raised KeyError:', e)

    done = [f.is_done for f in futures]
    print('  futures done:', done)
    accumulator.begin_flush()
    try:
        accumulator.await_flush_completion(timeout=1)  # what KafkaProducer.flush() does
        print('  flush: completed')
    except Errors.KafkaTimeoutError as e:
        print('  flush:', repr(e), '-> with timeout=None this blocks forever')
    return done


print('kafka-python', kafka.__version__)
print('as released:')
assert run() == [True, False]
print('with muted.remove -> discard:')
assert run(type('DiscardSet', (set,), {'remove': set.discard})()) == [True, True]
print('OK: bug reproduced, discard fixes it')

Output on 3.0.11 and on master:

as released:
  _send_producer_data raised KeyError: TopicPartition(topic='foo', partition=0)
  futures done: [True, False]
  flush: KafkaTimeoutError('Timeout waiting for future') -> with timeout=None this blocks forever
with muted.remove -> discard:
  futures done: [True, True]
  flush: completed
OK: bug reproduced, discard fixes it

What we saw in production

Four producer processes (acks='all', retries=3, other settings default) calling flush() after each small batch of send()s. During a network outage to the brokers, one process logged 147 of these KeyErrors. Most were harmless because only one batch expired in that pass. Each process eventually hit a pass that expired several batches at once. After that, 2 to 3 of the 12 messages in the pending flush() never received a callback or errback. Broker connectivity came back (metadata refreshes have succeeded every 5 minutes since), but flush() has not returned in over 37 hours.

Log excerpt (topic name anonymized):

WARNING - kafka.producer.producer_batch - Failed to produce messages to topic-partition TopicPartition(topic='topic-a', partition=5) with base offset None: KafkaTimeoutError: Expiring 1 record(s) for TopicPartition(topic='topic-a', partition=5): 120001 ms has passed since batch creation
ERROR - kafka.producer.sender - <Sender client_id=kafka-python-producer-1 transactional_id=None>: Uncaught error in kafka producer I/O thread
Traceback (most recent call last):
  File ".../kafka/producer/sender.py", line 113, in run
    self.run_once()
  File ".../kafka/producer/sender.py", line 187, in run_once
    poll_timeout_ms = self._send_producer_data()
  File ".../kafka/producer/sender.py", line 247, in _send_producer_data
    self._complete_batch_with_exception(expired_batch, Errors.KafkaTimeoutError(error_message))
  File ".../kafka/producer/sender.py", line 488, in _complete_batch_with_exception
    self._dispatch_error(batch, exception, partition_response=None)
  File ".../kafka/producer/sender.py", line 588, in _dispatch_error
    self._accumulator.muted.remove(batch.topic_partition)
KeyError: TopicPartition(topic='topic-a', partition=5)

Suggested fix

• Use self._accumulator.muted.discard(batch.topic_partition) in _dispatch_error, and in the success path of _complete_batch for consistency. This matches the remove-if-present handling already used in RecordAccumulator.abort_batches. The reproduction passes with this change.
• Consider making the expiry loop in _send_producer_data robust to an exception on one batch, so a single failure cannot orphan the rest of the list.

Happy to open a PR with the discard change and a regression test based on the script above.

Workaround

enable_idempotence=False (with max_in_flight_requests_per_connection above 1) skips all muted handling. Passing a timeout to flush() turns the hang into a KafkaTimeoutError, but the orphaned batches stay in the accumulator's incomplete set, so every later flush() times out as well until the producer is replaced.