काफ्का से प्राप्त घटनाओं का पुनर्प्रसंस्करण

हे हबर।

हाल ही में, मैं अपना अनुभव साझा किया इस बारे में कि एक टीम के रूप में हम काफ्का निर्माता और उपभोक्ता के लिए गारंटीकृत डिलीवरी के करीब पहुंचने के लिए किन मापदंडों का सबसे अधिक उपयोग करते हैं। इस लेख में मैं आपको बताना चाहता हूं कि बाहरी सिस्टम की अस्थायी अनुपलब्धता के परिणामस्वरूप हमने काफ्का से प्राप्त एक घटना के पुन: प्रसंस्करण का आयोजन कैसे किया।

आधुनिक अनुप्रयोग बहुत जटिल वातावरण में कार्य करते हैं। आधुनिक प्रौद्योगिकी स्टैक में लिपटे व्यावसायिक तर्क, कुबेरनेट्स या ओपनशिफ्ट जैसे ऑर्केस्ट्रेटर द्वारा प्रबंधित डॉकर छवि में चल रहे हैं, और भौतिक और आभासी राउटर की श्रृंखला के माध्यम से अन्य अनुप्रयोगों या एंटरप्राइज़ समाधानों के साथ संचार कर रहे हैं। ऐसे माहौल में, कुछ हमेशा टूट सकता है, इसलिए यदि कोई बाहरी सिस्टम अनुपलब्ध है तो घटनाओं को पुन: संसाधित करना हमारी व्यावसायिक प्रक्रियाओं का एक महत्वपूर्ण हिस्सा है।

काफ्का से पहले यह कैसा था

इससे पहले प्रोजेक्ट में हमने एसिंक्रोनस संदेश डिलीवरी के लिए आईबीएम एमक्यू का उपयोग किया था। यदि सेवा के संचालन के दौरान कोई त्रुटि हुई, तो प्राप्त संदेश को आगे की मैन्युअल पार्सिंग के लिए डेड-लेटर-क्यू (डीएलक्यू) में रखा जा सकता है। आने वाली कतार के बगल में DLQ बनाया गया था, संदेश को IBM MQ के अंदर स्थानांतरित किया गया था।

यदि त्रुटि अस्थायी थी और हम इसे निर्धारित कर सकते थे (उदाहरण के लिए, HTTP कॉल पर एक रिसोर्स एक्सेस अपवाद या MongoDb अनुरोध पर एक MongoTimeoutException), तो पुनः प्रयास रणनीति प्रभावी होगी। एप्लिकेशन के ब्रांचिंग तर्क के बावजूद, मूल संदेश को देरी से भेजने के लिए या तो सिस्टम कतार में ले जाया गया था, या संदेशों को फिर से भेजने के लिए बहुत पहले बनाए गए एक अलग एप्लिकेशन में ले जाया गया था। इसमें संदेश हेडर में एक पुनः भेजें नंबर शामिल है, जो विलंब अंतराल या एप्लिकेशन-स्तरीय रणनीति के अंत से जुड़ा हुआ है। यदि हम रणनीति के अंत तक पहुंच गए हैं लेकिन बाहरी सिस्टम अभी भी अनुपलब्ध है, तो संदेश को मैन्युअल पार्सिंग के लिए DLQ में रखा जाएगा।

समाधान के लिए खोजें

इंटरनेट पर खोज रहे हैं, आप निम्नलिखित पा सकते हैं निर्णय. संक्षेप में, प्रत्येक विलंब अंतराल के लिए एक विषय बनाने और उपभोक्ता अनुप्रयोगों को लागू करने का प्रस्ताव है, जो आवश्यक विलंब के साथ संदेशों को पढ़ेगा।

बड़ी संख्या में सकारात्मक समीक्षाओं के बावजूद, मुझे ऐसा लगता है कि यह पूरी तरह से सफल नहीं है। सबसे पहले, क्योंकि डेवलपर को, व्यावसायिक आवश्यकताओं को लागू करने के अलावा, वर्णित तंत्र को लागू करने में बहुत समय व्यतीत करना होगा।

इसके अलावा, यदि काफ्का क्लस्टर पर पहुंच नियंत्रण सक्षम है, तो आपको विषय बनाने और उन तक आवश्यक पहुंच प्रदान करने में कुछ समय बिताना होगा। इसके अतिरिक्त, आपको प्रत्येक पुनः प्रयास विषय के लिए सही retention.ms पैरामीटर का चयन करना होगा ताकि संदेशों को पुनः भेजे जाने का समय मिल सके और वे इससे गायब न हों। प्रत्येक मौजूदा या नई सेवा के लिए कार्यान्वयन और पहुंच के अनुरोध को दोहराना होगा।

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

उदाहरण के लिए, स्प्रिंग-काफ्का स्वयं कंटेनरअवेयरएररहैंडलर के कई कार्यान्वयन प्रदान करता है SeekToCurrentErrorHandler, जिसके साथ आप त्रुटि के मामले में ऑफ़सेट को स्थानांतरित किए बिना संदेश को बाद में संसाधित कर सकते हैं। स्प्रिंग-काफ्का 2.3 के संस्करण से शुरू करके, BackOffPolicy सेट करना संभव हो गया।

यह दृष्टिकोण पुन:संसाधित संदेशों को एप्लिकेशन पुनरारंभ से बचे रहने की अनुमति देता है, लेकिन अभी भी कोई DLQ तंत्र नहीं है। हमने 2019 की शुरुआत में इस विकल्प को चुना, आशावादी रूप से यह मानते हुए कि DLQ की आवश्यकता नहीं होगी (हम भाग्यशाली थे और वास्तव में इस तरह के पुनर्प्रसंस्करण प्रणाली के साथ एप्लिकेशन को संचालित करने के कई महीनों के बाद हमें इसकी आवश्यकता नहीं थी)। अस्थायी त्रुटियों के कारण SeekToCurrentErrorHandler सक्रिय हो गया। शेष त्रुटियाँ लॉग में मुद्रित की गईं, जिसके परिणामस्वरूप ऑफसेट हुआ, और अगले संदेश के साथ प्रसंस्करण जारी रहा।

अंतिम निर्णय

SeekToCurrentErrorHandler पर आधारित कार्यान्वयन ने हमें संदेशों को पुनः भेजने के लिए अपना स्वयं का तंत्र विकसित करने के लिए प्रेरित किया।

सबसे पहले, हम मौजूदा अनुभव का उपयोग करना चाहते थे और एप्लिकेशन लॉजिक के आधार पर इसका विस्तार करना चाहते थे। एक रैखिक तर्क अनुप्रयोग के लिए, पुनः प्रयास रणनीति द्वारा निर्दिष्ट थोड़े समय के लिए नए संदेशों को पढ़ना बंद करना इष्टतम होगा। अन्य अनुप्रयोगों के लिए, मैं एक ऐसा बिंदु चाहता था जो पुनः प्रयास रणनीति को लागू करेगा। इसके अलावा, इस एकल बिंदु में दोनों दृष्टिकोणों के लिए DLQ कार्यक्षमता होनी चाहिए।

पुन: प्रयास रणनीति को स्वयं एप्लिकेशन में संग्रहीत किया जाना चाहिए, जो अस्थायी त्रुटि होने पर अगले अंतराल को पुनर्प्राप्त करने के लिए जिम्मेदार है।

एक रेखीय तर्क अनुप्रयोग के लिए उपभोक्ता को रोकना

स्प्रिंग-काफ्का के साथ काम करते समय, उपभोक्ता को रोकने का कोड कुछ इस तरह दिख सकता है:

public void pauseListenerContainer(MessageListenerContainer listenerContainer, 
                                   Instant retryAt) {
        if (nonNull(retryAt) && listenerContainer.isRunning()) {
            listenerContainer.stop();
            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
            return;
        }
        // to DLQ
    }

उदाहरण में, यदि MessageListenerContainer अभी भी चल रहा है तो retryAt उसे पुनरारंभ करने का समय है। पुन: लॉन्च टास्क शेड्यूलर में लॉन्च किए गए एक अलग थ्रेड में होगा, जिसका कार्यान्वयन भी स्प्रिंग द्वारा प्रदान किया गया है।

हम निम्नलिखित तरीके से retryAt मान पाते हैं:

1. री-कॉल काउंटर का मूल्य देखा जाता है।
2. काउंटर वैल्यू के आधार पर, पुनः प्रयास रणनीति में वर्तमान विलंब अंतराल की खोज की जाती है। रणनीति एप्लिकेशन में ही घोषित की गई है; हमने इसे संग्रहीत करने के लिए JSON प्रारूप चुना है।
3. JSON सरणी में पाए गए अंतराल में सेकंड की संख्या होती है जिसके बाद प्रसंस्करण को दोहराने की आवश्यकता होगी। retryAt का मान बनाने के लिए सेकंड की यह संख्या वर्तमान समय में जोड़ी जाती है।
4. यदि अंतराल नहीं मिलता है, तो retryAt का मान शून्य है और संदेश मैन्युअल पार्सिंग के लिए DLQ को भेजा जाएगा।

इस दृष्टिकोण के साथ, जो कुछ बचा है वह वर्तमान में संसाधित किए जा रहे प्रत्येक संदेश के लिए बार-बार कॉल की संख्या को सहेजना है, उदाहरण के लिए एप्लिकेशन मेमोरी में। इस दृष्टिकोण के लिए स्मृति में पुनः प्रयास की गिनती को बनाए रखना महत्वपूर्ण नहीं है, क्योंकि एक रैखिक तर्क अनुप्रयोग संपूर्ण रूप से प्रसंस्करण को संभाल नहीं सकता है। स्प्रिंग-रिट्री के विपरीत, एप्लिकेशन को पुनरारंभ करने से खोए हुए सभी संदेशों को पुन: संसाधित नहीं किया जाएगा, बल्कि रणनीति पुनः आरंभ हो जाएगी।

यह दृष्टिकोण बाहरी सिस्टम से भार हटाने में मदद करता है, जो बहुत भारी भार के कारण अनुपलब्ध हो सकता है। दूसरे शब्दों में, पुनर्प्रसंस्करण के अलावा, हमने पैटर्न के कार्यान्वयन को हासिल किया सर्किट ब्रेकर.

हमारे मामले में, त्रुटि सीमा केवल 1 है, और अस्थायी नेटवर्क आउटेज के कारण सिस्टम डाउनटाइम को कम करने के लिए, हम छोटे विलंबता अंतराल के साथ एक बहुत ही विस्तृत पुनः प्रयास रणनीति का उपयोग करते हैं। यह सभी समूह अनुप्रयोगों के लिए उपयुक्त नहीं हो सकता है, इसलिए सिस्टम की विशेषताओं के आधार पर त्रुटि सीमा और अंतराल मान के बीच संबंध का चयन किया जाना चाहिए।

गैर-नियतात्मक तर्क वाले अनुप्रयोगों से संदेशों को संसाधित करने के लिए एक अलग एप्लिकेशन

यहां कोड का एक उदाहरण दिया गया है जो ऐसे एप्लिकेशन (Retryer) को एक संदेश भेजता है, जो RETRY_AT समय पहुंचने पर DESTINATION विषय पर फिर से भेजा जाएगा:

public <K, V> void retry(ConsumerRecord<K, V> record, String retryToTopic, 
                         Instant retryAt, String counter, String groupId, Exception e) {
        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
        List<Header> arrayOfHeaders = 
            new ArrayList<>(Arrays.asList(headers.toArray()));
        updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);
        updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);
        updateHeader(arrayOfHeaders, ORIGINAL_PARTITION, 
                     () -> Integer.toString(record.partition()).getBytes());
        if (nonNull(retryAt)) {
            updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);
            updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);
            updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);
        } else {
            updateHeader(arrayOfHeaders, REASON, 
                         ExceptionUtils.getStackTrace(e)::getBytes);
            updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);
        }
        ProducerRecord<K, V> messageToSend =
            new ProducerRecord<>(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
        kafkaTemplate.send(messageToSend);
    }

उदाहरण से पता चलता है कि हेडर में बहुत सारी जानकारी प्रसारित होती है। RETRY_AT का मान उसी तरह पाया जाता है जैसे उपभोक्ता स्टॉप के माध्यम से पुनः प्रयास तंत्र के लिए। DESTINATION और RETRY_AT के अलावा हम पास होते हैं:

• GROUP_ID, जिसके द्वारा हम मैन्युअल विश्लेषण और सरलीकृत खोज के लिए संदेशों को समूहित करते हैं।
• पुन: प्रसंस्करण के लिए उसी उपभोक्ता को रखने का प्रयास करने के लिए ORIGINAL_PARTITION। यह पैरामीटर शून्य हो सकता है, ऐसी स्थिति में मूल संदेश की रिकॉर्ड.की() कुंजी का उपयोग करके नया विभाजन प्राप्त किया जाएगा।
• पुनः प्रयास रणनीति का पालन करने के लिए अद्यतन COUNTER मान।
• SEND_TO एक स्थिरांक है जो दर्शाता है कि संदेश RETRY_AT तक पहुंचने पर पुन: प्रसंस्करण के लिए भेजा गया है या DLQ में रखा गया है।
• कारण - संदेश प्रसंस्करण बाधित होने का कारण।

पुनः प्रयासकर्ता संदेशों को पुनः भेजने और मैन्युअल पार्सिंग के लिए PostgreSQL में संग्रहीत करता है। एक टाइमर एक कार्य शुरू करता है जो RETRY_AT के साथ संदेश ढूंढता है और उन्हें कुंजी रिकॉर्ड.की() के साथ DESTINATION विषय के ORIGINAL_PARTITION विभाजन पर वापस भेजता है।

एक बार भेजे जाने के बाद, संदेश PostgreSQL से हटा दिए जाते हैं। संदेशों की मैन्युअल पार्सिंग एक साधारण यूआई में होती है जो REST API के माध्यम से रिट्रीयर के साथ इंटरैक्ट करती है। इसकी मुख्य विशेषताएं डीएलक्यू से संदेशों को दोबारा भेजना या हटाना, त्रुटि जानकारी देखना और संदेशों की खोज करना है, उदाहरण के लिए त्रुटि नाम से।

चूँकि हमारे क्लस्टर्स पर एक्सेस कंट्रोल सक्षम है, इसलिए अतिरिक्त रूप से उस विषय तक पहुंच का अनुरोध करना आवश्यक है जिसे रिट्रीयर सुन रहा है, और रिट्रीयर को DESTINATION विषय पर लिखने की अनुमति दें। यह असुविधाजनक है, लेकिन, अंतराल विषय दृष्टिकोण के विपरीत, हमारे पास इसे प्रबंधित करने के लिए एक पूर्ण डीएलक्यू और यूआई है।

ऐसे मामले होते हैं जब एक आने वाले विषय को कई अलग-अलग उपभोक्ता समूहों द्वारा पढ़ा जाता है, जिनके अनुप्रयोग अलग-अलग तर्क लागू करते हैं। इनमें से किसी एक एप्लिकेशन के लिए रिट्रीयर के माध्यम से किसी संदेश को दोबारा संसाधित करने पर दूसरे एप्लिकेशन पर डुप्लिकेट उत्पन्न हो जाएगा। इससे बचाव के लिए हम पुन: प्रसंस्करण के लिए एक अलग विषय बनाते हैं। आने वाले और पुनः प्रयास किए गए विषयों को एक ही उपभोक्ता बिना किसी प्रतिबंध के पढ़ सकता है।

डिफ़ॉल्ट रूप से यह दृष्टिकोण सर्किट ब्रेकर कार्यक्षमता प्रदान नहीं करता है, हालांकि इसका उपयोग करके एप्लिकेशन में जोड़ा जा सकता है स्प्रिंग-क्लाउड-नेटफ्लिक्स या नया स्प्रिंग क्लाउड सर्किट ब्रेकर, उन स्थानों को लपेटना जहां बाहरी सेवाओं को उचित अमूर्त में बुलाया जाता है। इसके अलावा, इसके लिए एक रणनीति चुनना संभव हो जाता है दिवार पैटर्न, जो उपयोगी भी हो सकता है। उदाहरण के लिए, स्प्रिंग-क्लाउड-नेटफ्लिक्स में यह थ्रेड पूल या सेमाफोर हो सकता है।

उत्पादन

परिणामस्वरूप, हमारे पास एक अलग एप्लिकेशन है जो हमें किसी बाहरी सिस्टम के अस्थायी रूप से अनुपलब्ध होने पर संदेश प्रसंस्करण को दोहराने की अनुमति देता है।

एप्लिकेशन के मुख्य लाभों में से एक यह है कि इसका उपयोग उसी काफ्का क्लस्टर पर चलने वाले बाहरी सिस्टम द्वारा किया जा सकता है, उनकी ओर से महत्वपूर्ण संशोधनों के बिना! ऐसे एप्लिकेशन को केवल पुनः प्रयास विषय तक पहुंचने, कुछ काफ्का हेडर भरने और पुनः प्रयासकर्ता को एक संदेश भेजने की आवश्यकता होगी। कोई अतिरिक्त बुनियादी ढांचा जुटाने की जरूरत नहीं है. और एप्लिकेशन से रिट्रीयर और वापस स्थानांतरित संदेशों की संख्या को कम करने के लिए, हमने रैखिक तर्क के साथ एप्लिकेशन की पहचान की और उपभोक्ता स्टॉप के माध्यम से उन्हें फिर से संसाधित किया।

स्रोत: www.habr.com