काफ्का निर्माता दलाल से कनेक्शन खो जाने से निपट रहा है

नीचे की तरह एक निर्माता विन्यास के साथ, मैं एक सिंगलटन निर्माता बना रहा हूं जिसका उपयोग पूरे एप्लिकेशन में किया जाता है:

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 होस्ट किए गए काफ्का क्लस्टर से जुड़ा हूं। ब्रोकर का advertised.listeners मुझे आईपी पते वापस करने के लिए कॉन्फ़िगर किया गया है न कि होस्ट नाम के लिए। जबकि आम तौर पर सब कुछ अपेक्षित रूप से काम करता है, समस्या तब होती है जब काफ्का को फिर से शुरू किया जाता है, कभी-कभी आईपी पता बदल जाता है। चूंकि निर्माता केवल पुराने आईपी के बारे में जानता है, वह संदेश भेजने के लिए उस होस्ट से जुड़ने की कोशिश करता रहता है और कोई भी संदेश नहीं जाता है।

मैं देखता हूं कि जब प्रेषण विफल हो जाता है तो एक 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);
                    }
                });

टाइमआउट अपवाद को अब कैसे संभाला जाना चाहिए? यह देखते हुए कि निर्माता को पूरे एप्लिकेशन में साझा किया गया है, कॉलबैक में बंद करना और फिर से बनाना सही नहीं लगता है।