🥇 काफ्काबाट प्राप्त घटनाहरू पुन: प्रशोधन गर्दै

हेर Habr.

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

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

काफ्का अघि कस्तो थियो

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

यदि त्रुटि अस्थायी थियो र हामीले यसलाई निर्धारण गर्न सक्छौं (उदाहरणका लागि, HTTP कलमा ResourceAccessException वा MongoDb अनुरोधमा MongoTimeoutException), त्यसपछि पुन: प्रयास रणनीति प्रभावकारी हुनेछ। एप्लिकेसनको ब्रान्चिङ तर्कको बावजुद, मूल सन्देश या त ढिलो पठाउनको लागि प्रणाली लाममा सारियो, वा सन्देशहरू पुन: पठाउनको लागि धेरै पहिले बनाइएको एउटा छुट्टै अनुप्रयोगमा सारिएको थियो। यसले सन्देश हेडरमा पुन: पठाउने नम्बर समावेश गर्दछ, जुन ढिलाइ अन्तराल वा अनुप्रयोग-स्तर रणनीतिको अन्त्यमा बाँधिएको हुन्छ। यदि हामी रणनीतिको अन्त्यमा पुगेका छौं तर बाह्य प्रणाली अझै उपलब्ध छैन भने, सन्देश म्यानुअल पार्सिङको लागि DLQ मा राखिनेछ।

समाधान खोज्दै

इन्टरनेटमा खोज्दै, तपाइँ निम्न पाउन सक्नुहुन्छ समाधान। छोटकरीमा, प्रत्येक ढिलाइ अन्तरालको लागि विषय सिर्जना गर्न र छेउमा उपभोक्ता अनुप्रयोगहरू लागू गर्ने प्रस्ताव गरिएको छ, जसले आवश्यक ढिलाइको साथ सन्देशहरू पढ्नेछ।

सकारात्मक समीक्षाहरूको ठूलो संख्याको बावजुद, यो पूर्णतया सफल छैन जस्तो लाग्छ। सबै भन्दा पहिले, किनभने विकासकर्ता, व्यापार आवश्यकताहरू लागू गर्न को लागी, वर्णन संयन्त्र लागू गर्न धेरै समय खर्च गर्नुपर्नेछ।

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

अब हेरौं सामान्य रूपमा वसन्त र विशेष रूपमा वसन्त-काफ्काले हामीलाई सन्देश पुन: प्रशोधनका लागि प्रदान गर्दछ। Spring-kafka को spring-retry मा एक ट्रान्जिटिभ निर्भरता छ, जसले विभिन्न BackOffPolis को प्रबन्ध गर्न को लागी abstractions प्रदान गर्दछ। यो एकदम लचिलो उपकरण हो, तर यसको महत्त्वपूर्ण कमजोरी भनेको अनुप्रयोग मेमोरीमा पुन: पठाउनका लागि सन्देशहरू भण्डारण गर्नु हो। यसको मतलब यो हो कि अपडेट वा अपरेशनल त्रुटिको कारणले अनुप्रयोगलाई पुन: सुरु गर्दा पुन: प्रशोधन गर्न बाँकी रहेका सबै सन्देशहरू हराउनेछ। यो बिन्दु हाम्रो प्रणालीको लागि महत्वपूर्ण भएकोले, हामीले यसलाई थप विचार गरेनौं।

spring-kafka आफैले ContainerAwareErrorHandler को धेरै कार्यान्वयनहरू प्रदान गर्दछ, उदाहरणका लागि SeekToCurrentErrorHandler, जसको साथ तपाईले सन्देशलाई पछि प्रशोधन गर्न सक्नुहुन्छ कुनै त्रुटिको मामलामा अफसेट सिफ्ट नगरी। Spring-kafka 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
    }

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

हामी निम्न तरिकामा retryAt मान फेला पार्छौं:

पुन: कल काउन्टरको मूल्य हेरिएको छ।
काउन्टर मानको आधारमा, पुन: प्रयास रणनीतिमा हालको ढिलाइ अन्तराल खोजिएको छ। रणनीति अनुप्रयोगमा नै घोषणा गरिएको छ; हामीले यसलाई भण्डारण गर्न JSON ढाँचा रोज्यौं।
JSON array मा पाइने अन्तरालले सेकेन्डको संख्या समावेश गर्दछ जुन पछि प्रक्रिया दोहोर्याउनु पर्छ। सेकेन्ड को यो संख्या retryAt को लागि मान बनाउन वर्तमान समयमा थपिएको छ।
यदि अन्तराल फेला परेन भने, retryAt को मान शून्य छ र सन्देश म्यानुअल पार्सिङको लागि DLQ मा पठाइनेछ।

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

यो दृष्टिकोणले बाह्य प्रणालीबाट भार हटाउन मद्दत गर्दछ, जुन धेरै भारी भारको कारण अनुपलब्ध हुन सक्छ। अर्को शब्दमा, पुन: प्रशोधनको अतिरिक्त, हामीले ढाँचाको कार्यान्वयन हासिल गर्यौं सर्किट ब्रेकर.

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

गैर-निर्धारित तर्कको साथ अनुप्रयोगहरूबाट सन्देशहरू प्रशोधन गर्नको लागि छुट्टै अनुप्रयोग

RETRY_AT समय पुगेपछि DESTINATION शीर्षकमा पुन: पठाउने एउटा यस्तो अनुप्रयोग (Retryer) मा सन्देश पठाउने कोडको उदाहरण यहाँ छ:


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 मा राखिएको हो भनेर संकेत गर्ने स्थिर हो।
REASON - सन्देश प्रशोधन अवरुद्ध भएको कारण।

पुन: प्रयासकर्ताले PostgreSQL मा पुन: पठाउने र म्यानुअल पार्सिङका लागि सन्देशहरू भण्डार गर्दछ। टाइमरले एउटा कार्य सुरु गर्छ जसले RETRY_AT सँग सन्देशहरू फेला पार्छ र तिनीहरूलाई कुञ्जी record.key() को साथ DESTINATION विषयको ORIGINAL_PARTITION विभाजनमा फिर्ता पठाउँछ।

एक पटक पठाइएपछि, सन्देशहरू PostgreSQL बाट मेटाइन्छ। सन्देशहरूको म्यानुअल पार्सिङ एक साधारण UI मा हुन्छ जसले REST API मार्फत Retryer सँग अन्तरक्रिया गर्छ। यसको मुख्य विशेषताहरू DLQ बाट सन्देशहरू पुन: पठाउने वा मेटाउने, त्रुटि जानकारी हेर्ने र सन्देशहरू खोज्ने, उदाहरणका लागि त्रुटि नामद्वारा।

हाम्रो क्लस्टरहरूमा पहुँच नियन्त्रण सक्षम गरिएको हुनाले, पुन: प्रयासकर्ताले सुनिरहेको विषयमा पहुँचको लागि अनुरोध गर्न आवश्यक छ, र पुन: प्रयासकर्तालाई DESTINATION शीर्षकमा लेख्न अनुमति दिनुहोस्। यो असुविधाजनक छ, तर, अन्तराल विषय दृष्टिकोणको विपरीत, हामीसँग यसलाई व्यवस्थित गर्नको लागि पूर्ण DLQ र UI छ।

त्यहाँ केसहरू छन् जब आगमन विषय धेरै फरक उपभोक्ता समूहहरू द्वारा पढिन्छ, जसको अनुप्रयोगहरूले फरक तर्क लागू गर्दछ। यी अनुप्रयोगहरू मध्ये एउटाको लागि Retryer मार्फत सन्देश पुन: प्रशोधन गर्दा अर्कोमा नक्कल हुनेछ। यसबाट जोगाउन, हामी पुन: प्रशोधनको लागि छुट्टै विषय सिर्जना गर्छौं। आगमन र पुन: प्रयास विषयहरू कुनै पनि प्रतिबन्ध बिना उही उपभोक्ता द्वारा पढ्न सकिन्छ।

पूर्वनिर्धारित रूपमा यो दृष्टिकोणले सर्किट ब्रेकर कार्यक्षमता प्रदान गर्दैन, यद्यपि यो प्रयोग गरेर अनुप्रयोगमा थप्न सकिन्छ वसन्त-क्लाउड-नेटफ्लिक्स वा नयाँ वसन्त क्लाउड सर्किट ब्रेकर, बाह्य सेवाहरूलाई उपयुक्त एब्स्ट्र्याक्सनमा बोलाइने ठाउँहरू बेर्दै। साथै, यो को लागी एक रणनीति छनोट गर्न सम्भव हुन्छ बल्कहेड ढाँचा, जुन उपयोगी पनि हुन सक्छ। उदाहरणका लागि, वसन्त-क्लाउड-नेटफ्लिक्समा यो थ्रेड पूल वा सेमाफोर हुन सक्छ।

निष्कर्षमा

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

एप्लिकेसनको मुख्य फाइदाहरू मध्ये एक यो हो कि यो समान काफ्का क्लस्टरमा चलिरहेको बाह्य प्रणालीहरू द्वारा प्रयोग गर्न सकिन्छ, तिनीहरूको पक्षमा महत्त्वपूर्ण परिमार्जनहरू बिना! यस्तो एप्लिकेसनले मात्र पुन: प्रयास शीर्षकमा पहुँच गर्न आवश्यक छ, केहि काफ्का हेडरहरू भर्नुहोस् र पुन: प्रयासकर्तालाई सन्देश पठाउनुहोस्। कुनै थप पूर्वाधार बढाउन आवश्यक छैन। र एप्लिकेसनबाट Retryer र ब्याकमा स्थानान्तरण गरिएका सन्देशहरूको संख्या घटाउनको लागि, हामीले रैखिक तर्कका साथ अनुप्रयोगहरू पहिचान गर्यौं र तिनीहरूलाई उपभोक्ता स्टप मार्फत पुन: प्रशोधन गर्यौं।

स्रोत: www.habr.com

काफ्काबाट प्राप्त घटनाहरू पुन: प्रशोधन गर्दै