title: Kafka producer dealing with lost connection to broker
body: <p>With a producer configuration like below, I am creating a Singleton producer that is used throughout the application:</p>
  <pre><code>properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;kafka.consul1:9092,kafka.consul2:9092&quot;);
  properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  properties.setProperty(ProducerConfig.ACKS_CONFIG, &quot;1&quot;);
  
  </code></pre>
  <p>I am connected to a k8s hosted Kafka cluster. The broker's <code>advertised.listeners</code> is configured to return me the IP addresses and not host names. While normally everything works as expected, the problem occurs when Kafka is restarted, sometimes the IP address changes. Since the producer only knows about the older IP it keeps trying to connect to that host to send messages and none of the messages go through.</p>
  <p>I observe that a <code> org.apache.kafka.common.errors.TimeoutException</code> exception is thrown when the send fails. Currently the messages are sent async:</p>
  <pre><code>producer.send(data,
                  (RecordMetadata recordMetadata, Exception e) -&gt; {
                      if (e != null) {
                          LOGGER.error(&quot;Exception while sending message to kafka&quot;, e);
                      }
                  });
  
  </code></pre>
  <p>How should the Timeoutexception be handled now? Given that the producer is shared across the application, closing and recreating in the callback does not sound right.</p>