Missing retry backoff in Fetcher for NotLeaderForPartitionError causes tight fetch loop during broker restarts

[vanchaxy]

When Fetcher._proc_fetch_request receives NotLeaderForPartitionError or UnknownTopicOrPartitionError in the per-partition response loop, it calls self._client.force_metadata_update() but does not apply any backoff/sleep before the next fetch cycle.

Normally Kafka broker waits up to fetch.max.wait.ms (default 500ms) before responding to a fetch request. However, when a partition has no leader (e.g. during broker restart or partition reassignment), the broker returns NotLeaderForPartitionError immediately without entering the wait phase. The response time drops to just the network round-trip (~10-20ms).

Without a backoff on the client side, the fetch loop spins as fast as the network allows, effectively DDoS-ing the Kafka cluster with fetch requests. This causes memory usage to spike in the consumer process.

Note that the general KafkaError exception handler at the top of the method (added in #534) does include await asyncio.sleep(self._retry_backoff), but the per-partition error handling branch for NotLeaderForPartitionError inside the response processing loop does not.

Expected behaviour

After receiving NotLeaderForPartitionError or UnknownTopicOrPartitionError, the fetcher should apply self._retry_backoff sleep (same as it does for general KafkaError), giving the cluster time to elect new partition lleaders without being overwhelmed by requests.

  elif error_type in (
      Errors.NotLeaderForPartitionError,
      Errors.UnknownTopicOrPartitionError,
  ):
      self._client.force_metadata_update()
      await asyncio.sleep(self._retry_backoff)  # missing backoff

[vmaurin]

_proc_fetch_request should be called only with a node id. If the metadata are refreshed, then the client should "know" there are no leaders for the partition and then don't call _proc_fetch_request again. Do you have a debug log extract showing the behavior you are describing ?

[vanchaxy]

kinda... I have some screenshots of logs from version 0.10.0 from testing 2 years ago. We fixed it by patching aiokafka with sleep, and I forgot to report it here.

Kafka with 3 nodes and a topic with 64 partitions. There are logs after restarting kafka (hard restart by dropping kafka pods).

During 30s window 33k "adding fetch requests" logs.

During this 30s window aiokafka made 830 requests like these. All requests with raplica_id=-1:

And received 830 responses like these. Some partitions with offset and error_code=0, and others with error_code=6:

Time between request and response in ms:

Time between request and response during normal operation:

[vmaurin]

Replica id = -1 is "normal", what you can see from your log is the connection to the kafka-1 broker, that then it is probably believed to be the leader of the queried partition. If I understand well your setup, you have a replication factor of 1 right ? (so if a broker is down, there are no leaders for a bunch of partitions)

[vanchaxy]

This is our kafka config:

      min.insync.replicas: 2
      default.replication.factor: 3
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2

If I recall correctly, the problem is that aiokafka is trying to refresh metadata from a random node, and sometimes it chooses one that has not yet started. Metadata refresh have timeout of 40s, and during these 40s, we DDoS Kafka with fetch requests, and other (already started) nodes respond to us instantly. Does that make any sense?

[vmaurin]

Ok that makes sense (i.e picking the down node then having a timeout). So the backoff could be used, otherwise maybe the fetcher should wait for the metadata refresh to complete ? In practice it doesn't seem a good idea to try to fetch again the same partitions if we don't get a clear picture of what is going on in the cluster.

Maybe we could take a look of the behavior of the java client in such a scenario

[vanchaxy]

the fetcher should wait for the metadata refresh to complete

Can this be done by calling await self._client._maybe_wait_metadata() before calling _get_actions_per_node?