trying to handle broker disconnection in KafkaConsumer