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 ?

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

@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?

@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.

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

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.

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

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.

I like of idea of building a heartbeat mechanism using the listTopics method, as @egast suggested.

I don't think it's always desirable to fail the stream when the consumer cannot connect for some period. In long running applications you may just want to wait until the broker is back online. But it could also signal that you have a wrong configration and the connection cannot be established because of that, in which case it would be good to fail fast. There's also errors that can occur for a specific subscription, i.e. when not authorized for some topic. I don't know if we can detect those and fail the stream, while potentially keeping streams for other subscriptions running.

In any case we could have a diagnostic event Connected / Disconnected, which would allow the user to decide how to handle connection issues. A new metric 'connection state', like we have for subscription state would also allow for monitoring and alerting.