I want to stop when Kafka throws and error and resume sending when it recovers from the error. For example if the cluster is down, I wont sent messages but I want to start automatically when connection is established. How can I get the state information of the producer?