Question: Throttle Reconnect Attempts when kafka is unavailable

[jaredvacanti]

Checklist

• I have included information about relevant versions
• I have verified that the issue persists when using the master branch of Faust.

Steps to reproduce

Tell us what you did to cause something to happen.

• Connect Faust App normally to a kafka broker
• Kill kafka process

Expected behavior

The reconnect attempts aren't really unexpected. Is there a way to throttle the reconnect attempts or stop the logs from filling so rapidly? I can't find how to pass configuration to the aiokafka transport, which I think can accept reconnect.backoff.ms.

Actual behavior

The logs are spam with Errors very quickly. It looks like 100ms delay.

Full traceback

◠[2024-04-26 13:04:28,809] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◝[2024-04-26 13:04:28,811] [35601] [ERROR] Unable to update metadata from [1] 
◝[2024-04-26 13:04:28,852] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◞[2024-04-26 13:04:28,853] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◝[2024-04-26 13:04:28,869] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◞[2024-04-26 13:04:28,870] [35601] [ERROR] Unable to update metadata from [1] 
◞[2024-04-26 13:04:28,976] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◡[2024-04-26 13:04:28,977] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◞[2024-04-26 13:04:28,982] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◡[2024-04-26 13:04:28,983] [35601] [ERROR] Unable to update metadata from [1] 
◡[2024-04-26 13:04:29,089] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◟[2024-04-26 13:04:29,090] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◡[2024-04-26 13:04:29,098] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◟[2024-04-26 13:04:29,098] [35601] [ERROR] Unable to update metadata from [1] 
◟[2024-04-26 13:04:29,204] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◜[2024-04-26 13:04:29,204] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◟[2024-04-26 13:04:29,209] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◜[2024-04-26 13:04:29,210] [35601] [ERROR] Unable to update metadata from [1] 
◜[2024-04-26 13:04:29,316] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◠[2024-04-26 13:04:29,317] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◜[2024-04-26 13:04:29,327] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◠[2024-04-26 13:04:29,328] [35601] [ERROR] Unable to update metadata from [1] 
◝[2024-04-26 13:04:29,435] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◞[2024-04-26 13:04:29,436] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◝[2024-04-26 13:04:29,441] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◞[2024-04-26 13:04:29,441] [35601] [ERROR] Unable to update metadata from [1] 
◡[2024-04-26 13:04:29,546] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◟[2024-04-26 13:04:29,546] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◡[2024-04-26 13:04:29,551] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◟[2024-04-26 13:04:29,551] [35601] [ERROR] Unable to update metadata from [1] 
◟[2024-04-26 13:04:29,656] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◜[2024-04-26 13:04:29,656] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◟[2024-04-26 13:04:29,660] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◜[2024-04-26 13:04:29,661] [35601] [ERROR] Unable to update metadata from [1] 
◜[2024-04-26 13:04:29,766] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◠[2024-04-26 13:04:29,767] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◜[2024-04-26 13:04:29,776] [35601] [ERROR] Unable connect to node with id 1: KafkaConnectionError: Connection at localhost:9092 closed 
◠[2024-04-26 13:04:29,777] [35601] [ERROR] Unable to update metadata from [1] 
◡[2024-04-26 13:04:29,881] [35601] [ERROR] Unable connect to node with id 1: Multiple exceptions: [Errno 61] Connect call failed ('::1', 9092, 0, 0), [Errno 61] Connect call failed ('127.0.0.1', 9092) 
◟[2024-04-26 13:04:29,882] [35601] [ERROR] Group Coordinator Request failed: NodeNotReadyError: Attempt to send a request to node which is not ready (node id 1). 
◠[2024-04-26 13:04:29,887] [35601] [ERROR] Unable connect to node with id 1: Multiple exceptions: [Errno 61] Connect call failed ('::1', 9092, 0, 0), [Errno 61] Connect call failed ('127.0.0.1', 9092) ```

# Versions

* Python version Cython 3.11.8
* Faust version v0.11.0
* Operating system Darwin x86_64
* Kafka version 3.7.0
* RocksDB version (if applicable) 9.0.0

[jaredvacanti]

I found these issues from aiokafka that may be relevant here:

• Too many reqs on kafka not available  aio-libs/aiokafka#496
• catch "NodeNotReadyError" exception while consuming from Kafka aio-libs/aiokafka#612
• Use sleep when raise exception asyncio.TimeoutError, KafkaError aio-libs/aiokafka#949

I'm still not sure if there's a way to pass the reconnect_backoff_ms as a config option or somehow catch & handle the KafkaConnectionError in faust.