काफ्काकडून प्राप्त झालेल्या घटनांची पुनर्प्रक्रिया करणे

अहो हब्र.

अलीकडे आय त्याचा अनुभव शेअर केला गॅरंटीड डिलिव्हरीच्या जवळ जाण्यासाठी आम्ही काफ्का उत्पादक आणि ग्राहक यांच्यासाठी टीम म्हणून कोणते पॅरामीटर्स वापरतो. या लेखात मी तुम्हाला सांगू इच्छितो की आम्ही बाह्य प्रणालीच्या तात्पुरत्या अनुपलब्धतेचा परिणाम म्हणून काफ्काकडून प्राप्त झालेल्या कार्यक्रमाची पुनर्प्रक्रिया कशी आयोजित केली.

आधुनिक अनुप्रयोग अतिशय जटिल वातावरणात कार्य करतात. आधुनिक तंत्रज्ञानाच्या स्टॅकमध्ये गुंडाळलेले व्यवसाय लॉजिक, कुबर्नेट्स किंवा ओपनशिफ्ट सारख्या ऑर्केस्ट्रेटरद्वारे व्यवस्थापित केलेल्या डॉकर इमेजमध्ये चालत आहे आणि भौतिक आणि व्हर्च्युअल राउटरच्या साखळीद्वारे इतर ऍप्लिकेशन्स किंवा एंटरप्राइझ सोल्यूशन्सशी संप्रेषण करत आहे. अशा वातावरणात, काहीतरी नेहमी खंडित होऊ शकते, त्यामुळे एखादी बाह्य प्रणाली अनुपलब्ध असल्यास घटनांची पुनर्प्रक्रिया करणे हा आमच्या व्यवसाय प्रक्रियेचा एक महत्त्वाचा भाग आहे.

काफ्काच्या आधी ते कसे होते

याआधी प्रोजेक्टमध्ये आम्ही असिंक्रोनस मेसेज डिलिव्हरीसाठी IBM MQ वापरला होता. सेवेच्या कार्यादरम्यान कोणतीही त्रुटी आढळल्यास, प्राप्त केलेला संदेश पुढील मॅन्युअल पार्सिंगसाठी डेड-लेटर-क्यू (DLQ) मध्ये ठेवला जाऊ शकतो. येणार्‍या रांगेच्या पुढे DLQ तयार केला होता, संदेश IBM MQ मध्ये हस्तांतरित केला गेला होता.

जर त्रुटी तात्पुरती असेल आणि आम्ही ती निश्चित करू शकलो (उदाहरणार्थ, HTTP कॉलवरील ResourceAccessException किंवा MongoDb विनंतीवर MongoTimeoutException), तर पुन्हा प्रयत्न करण्याचे धोरण प्रभावी होईल. ऍप्लिकेशनच्या ब्रँचिंग लॉजिकची पर्वा न करता, मूळ संदेश एकतर विलंबाने पाठवल्याबद्दल सिस्टम रांगेत हलविला गेला किंवा संदेश पुन्हा पाठवण्यासाठी खूप पूर्वी तयार केलेल्या वेगळ्या ऍप्लिकेशनमध्ये हलविला गेला. यामध्ये मेसेज हेडरमध्ये पुन्हा पाठवलेल्या नंबरचा समावेश आहे, जो विलंब मध्यांतर किंवा अनुप्रयोग-स्तर धोरणाच्या समाप्तीशी जोडलेला आहे. जर आम्ही रणनीतीच्या शेवटी पोहोचलो आहोत परंतु बाह्य प्रणाली अद्याप अनुपलब्ध आहे, तर मॅन्युअल पार्सिंगसाठी संदेश DLQ मध्ये ठेवला जाईल.

उपाय शोधणे

इंटरनेटवर शोधत आहे, आपण खालील शोधू शकता решение. थोडक्यात, प्रत्येक विलंब अंतरासाठी एक विषय तयार करणे आणि बाजूला ग्राहक अनुप्रयोग लागू करणे प्रस्तावित आहे, जे आवश्यक विलंबासह संदेश वाचतील.

मोठ्या संख्येने सकारात्मक पुनरावलोकने असूनही, मला असे वाटते की ते पूर्णपणे यशस्वी झाले नाही. सर्व प्रथम, कारण विकसकाला, व्यवसाय आवश्यकता लागू करण्याव्यतिरिक्त, वर्णन केलेल्या यंत्रणेची अंमलबजावणी करण्यासाठी बराच वेळ घालवावा लागेल.

याव्यतिरिक्त, जर काफ्का क्लस्टरवर प्रवेश नियंत्रण सक्षम केले असेल, तर तुम्हाला विषय तयार करण्यासाठी आणि त्यांना आवश्यक प्रवेश प्रदान करण्यासाठी थोडा वेळ द्यावा लागेल. या व्यतिरिक्त, तुम्हाला पुन्हा प्रयत्न करण्याच्या प्रत्येक विषयासाठी योग्य retention.ms पॅरामीटर निवडणे आवश्यक आहे जेणेकरून संदेशांना पुन्हा पाठवण्याची वेळ मिळेल आणि त्यातून अदृश्य होणार नाही. प्रत्येक विद्यमान किंवा नवीन सेवेसाठी अंमलबजावणी आणि प्रवेशाची विनंती पुनरावृत्ती करावी लागेल.

आता सामान्यतः स्प्रिंग आणि विशेषत: स्प्रिंग-काफ्का संदेशाच्या पुनर्प्रक्रियासाठी कोणती यंत्रणा पुरवतात ते पाहू. स्प्रिंग-काफ्कामध्ये स्प्रिंग-पुनर्प्रयासावर एक संक्रमणात्मक अवलंबित्व आहे, जे विविध बॅकऑफ पॉलिसी व्यवस्थापित करण्यासाठी अमूर्तता प्रदान करते. हे बर्‍यापैकी लवचिक साधन आहे, परंतु त्याचा महत्त्वपूर्ण दोष म्हणजे अनुप्रयोग मेमरीमध्ये पुन्हा पाठवण्यासाठी संदेश संचयित करणे. याचा अर्थ असा की अपडेट किंवा ऑपरेशनल एररमुळे अॅप्लिकेशन रीस्टार्ट केल्याने पुनर्प्रक्रिया प्रलंबित असलेले सर्व संदेश नष्ट होतील. हा मुद्दा आमच्या प्रणालीसाठी गंभीर असल्याने, आम्ही त्याचा अधिक विचार केला नाही.

spring-kafka स्वतः ContainerAwareErrorHandler ची अनेक अंमलबजावणी प्रदान करते, उदाहरणार्थ SeekToCurrentErrorHandler, ज्याद्वारे तुम्ही त्रुटीच्या बाबतीत ऑफसेट न हलवता नंतर संदेशावर प्रक्रिया करू शकता. स्प्रिंग-काफ्का 2.3 च्या आवृत्तीपासून सुरुवात करून, बॅकऑफ पॉलिसी सेट करणे शक्य झाले.

हा दृष्टीकोन रीप्रोसेस्ड संदेशांना ऍप्लिकेशन रीस्टार्ट होण्यासाठी टिकून राहण्याची परवानगी देतो, परंतु अद्याप कोणतीही 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 आहे आणि तात्पुरत्या नेटवर्क आउटेजमुळे सिस्टम डाउनटाइम कमी करण्यासाठी, आम्ही लहान लेटन्सी मध्यांतरांसह एक अतिशय बारीक पुन्हा प्रयत्न करण्याचे धोरण वापरतो. हे सर्व गट अनुप्रयोगांसाठी योग्य असू शकत नाही, म्हणून त्रुटी थ्रेशोल्ड आणि मध्यांतर मूल्य यांच्यातील संबंध सिस्टमच्या वैशिष्ट्यांवर आधारित निवडले जाणे आवश्यक आहे.

नॉन-डिटरमिनिस्टिक लॉजिकसह अनुप्रयोगांमधील संदेशांवर प्रक्रिया करण्यासाठी एक स्वतंत्र अनुप्रयोग

येथे कोडचे एक उदाहरण आहे जे अशा ऍप्लिकेशनला संदेश पाठवते (पुन्हा प्रयत्न करणार्‍या), जे 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 पुन्हा प्रक्रिया करण्यासाठी समान ग्राहक ठेवण्याचा प्रयत्न करण्यासाठी. हे पॅरामीटर शून्य असू शकते, अशा परिस्थितीत मूळ संदेशाची record.key() की वापरून नवीन विभाजन प्राप्त केले जाईल.
• पुन्हा प्रयत्न करण्याच्या धोरणाचे अनुसरण करण्यासाठी COUNTER मूल्य अद्यतनित केले.
• SEND_TO हा संदेश RETRY_AT वर पोहोचल्यावर पुनर्प्रक्रियेसाठी पाठवला गेला आहे किंवा DLQ मध्ये ठेवला आहे हे दर्शविते.
• कारण - संदेश प्रक्रियेत व्यत्यय येण्याचे कारण.

रीट्रायर पोस्टग्रेएसक्यूएलमध्ये पुन्हा पाठवण्यासाठी आणि मॅन्युअल पार्सिंगसाठी संदेश संग्रहित करते. एक टायमर एक कार्य सुरू करतो जो RETRY_AT सह संदेश शोधतो आणि की record.key() सह DESTINATION विषयाच्या ORIGINAL_PARTITION विभाजनावर परत पाठवतो.

एकदा पाठवल्यानंतर, संदेश PostgreSQL वरून हटवले जातात. संदेशांचे मॅन्युअल पार्सिंग एका साध्या UI मध्ये होते जे REST API द्वारे Retryer शी संवाद साधते. त्याची मुख्य वैशिष्ट्ये DLQ वरून संदेश पुन्हा पाठवणे किंवा हटवणे, त्रुटी माहिती पाहणे आणि संदेश शोधणे, उदाहरणार्थ त्रुटी नावाने.

आमच्या क्लस्टर्सवर ऍक्सेस कंट्रोल सक्षम केलेले असल्याने, रीट्रिअर ऐकत असलेल्या विषयावर ऍक्सेसची विनंती करणे आवश्यक आहे आणि रीट्रियरला DESTINATION विषयावर लिहिण्याची परवानगी देणे आवश्यक आहे. हे गैरसोयीचे आहे, परंतु, मध्यांतर विषयाच्या दृष्टिकोनाच्या विपरीत, ते व्यवस्थापित करण्यासाठी आमच्याकडे पूर्ण वाढ झालेला DLQ आणि UI आहे.

अशी प्रकरणे आहेत जेव्हा येणारे विषय अनेक भिन्न ग्राहक गटांद्वारे वाचले जातात, ज्यांचे अनुप्रयोग भिन्न तर्क लागू करतात. यापैकी एका अॅप्लिकेशनसाठी Retryer द्वारे मेसेजची पुनर्प्रक्रिया केल्यास दुसऱ्यावर डुप्लिकेट होईल. यापासून संरक्षण करण्यासाठी, आम्ही पुन्हा प्रक्रियेसाठी स्वतंत्र विषय तयार करतो. येणारे आणि पुन्हा प्रयत्न करणारे विषय समान ग्राहक कोणत्याही निर्बंधांशिवाय वाचू शकतात.

डीफॉल्टनुसार हा दृष्टिकोन सर्किट ब्रेकर कार्यक्षमता प्रदान करत नाही, तथापि ते वापरून ऍप्लिकेशनमध्ये जोडले जाऊ शकते स्प्रिंग-क्लाउड-नेटफ्लिक्स किंवा नवीन स्प्रिंग क्लाउड सर्किट ब्रेकर, ज्या ठिकाणी बाह्य सेवांना योग्य अ‍ॅबस्ट्रॅक्शनमध्ये बोलावले जाते त्या ठिकाणी गुंडाळणे. याव्यतिरिक्त, यासाठी एक धोरण निवडणे शक्य होते बल्कहेड नमुना, जे उपयुक्त देखील असू शकते. उदाहरणार्थ, स्प्रिंग-क्लाउड-नेटफ्लिक्समध्ये हे थ्रेड पूल किंवा सेमाफोर असू शकते.

निष्कर्ष

परिणामी, आमच्याकडे एक स्वतंत्र अनुप्रयोग आहे जो आम्हाला संदेश प्रक्रिया पुन्हा करण्याची परवानगी देतो जर कोणतीही बाह्य प्रणाली तात्पुरती अनुपलब्ध असेल.

ऍप्लिकेशनचा एक मुख्य फायदा असा आहे की ते एकाच काफ्का क्लस्टरवर चालणाऱ्या बाह्य प्रणालींद्वारे वापरले जाऊ शकते, त्यांच्या बाजूला लक्षणीय बदल न करता! अशा ऍप्लिकेशनला फक्त पुन्हा प्रयत्न करण्याच्या विषयावर प्रवेश करणे आवश्यक आहे, काही काफ्का शीर्षलेख भरा आणि पुन्हा प्रयत्न करणार्‍याला संदेश पाठवा. कोणत्याही अतिरिक्त पायाभूत सुविधा उभारण्याची गरज नाही. आणि ॲप्लिकेशनमधून Retryer आणि बॅककडे हस्तांतरित केलेल्या संदेशांची संख्या कमी करण्यासाठी, आम्ही रेखीय तर्कासह अनुप्रयोग ओळखले आणि ग्राहक स्टॉपद्वारे त्यांची पुन्हा प्रक्रिया केली.

स्त्रोत: www.habr.com