[QUESTION] Unable connect to node with id: X: [Errno 111]: Connection refused

[OlehDziuba]

Hi, I have group of consumers that on startup connect to kafka and infinite consume to messages. And I've encountered a problem that when the address of the kafka changes, the consumers don't get an error, just an infinite error logging. Since the Kafka address is taken from environment variables and I have an automatic reboot if an error occurs, it would be great if the consumer would raise an error

 consumer = AIOKafkaConsumer(
        *kafka_consumer_settings.topics
        bootstrap_servers=kafka_consumer_settings.bootstrap_servers,
        group_id=kafka_consumer_settings.group_id
    )
 
 try:
     async for msg in consumer:
         # Some message processing 
 finally:
     await consumer.stop()  # <--- it is not executed  in case of 'Unable connect to node with id: X: [Errno 111]: Connection refused'

I found it in source code:

aiokafka/aiokafka/client.py

Lines 475 to 481 in 256ce17

	except (OSError, asyncio.TimeoutError, KafkaError) as err:
	log.error("Unable connect to node with id %s: %s", node_id, err)
	if group == ConnectionGroup.DEFAULT:
	# Connection failures imply that our metadata is stale, so
	# let's refresh
	self.force_metadata_update()
	return None

Do I understand correctly that the error is not triggered only when the consumer is in a group?