使用下面这样的生产者配置，我将创建一个在整个应用程序中使用的单例生产者：

properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka.consul1:9092,kafka.consul2:9092");
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, "1");

我已连接到k8s托管的kafka群集。经纪人的 advertised.listeners 配置为返回ip地址而不是主机名。虽然正常情况下一切正常工作，但问题发生在Kafka重启时，有时ip地址会发生变化。由于生产者只知道旧的ip，它一直试图连接到该主机发送消息，没有消息通过。
我注意到 org.apache.kafka.common.errors.TimeoutException 发送失败时引发异常。当前消息是异步发送的：

producer.send(data,
                (RecordMetadata recordMetadata, Exception e) -> {
                    if (e != null) {
                        LOGGER.error("Exception while sending message to kafka", e);
                    }
                });

现在应该如何处理timeoutexception？由于生产者是在整个应用程序中共享的，因此在回调中关闭和重新创建听起来并不正确。