Close consumer when it cannot connect to the broker

[egast]

Add option to close the consumer when it is unable to connect to the broker. This should be configurable i.e. after some timeout and/or after x amount of retries.

[saraiva132]

I would very much love to see this! It is extremely annoying to have a consumer hang when it cannot connect to a broker.

WDYT @iravid ?

[svroonland]

As a workaround, have you considered a .timeout on the stream?

[svroonland]

@egast @saraiva132 Are you by any chance familiar which how to get connection status or something from the org.apache.kafka.clients.consumer.Consumer or some settings that control the behavior on connection timeout?

[egast]

@svroonland I don't know if you can get the Kafka connection status from a KafkaConsumer. I think the connection retry behavior is done by the consumer client. I know that Akka Kafka Streams does offer the "close consumer if it cannot connect to Kafka", so I took a quick look at how they do it and from what I understand they just periodically try to get the topic metadata to see if Kafka is still available. Maybe something similar can be implemented for zio-kafka.

[svroonland]

We could perhaps do something with the lack of a partitions assigned event from the RebalanceListener.

[erikvanoosten]

We could perhaps do something with the lack of a partitions assigned event from the RebalanceListener.

This would exclude use case that have extra instances standing by. This includes redeployments where you first start new instances, and then stop the old instances.

[svroonland]

Not sure, but do those instances get an empty list of assigned partitions perhaps?

[erikvanoosten]

Not sure, but do those instances get an empty list of assigned partitions perhaps?

Not sure either. It might be that the rebalance listener would not be invoked at all.