काफ्का कसरी वास्तविकता बन्यो

हे हाबर!

म Tinkoff टोलीमा काम गर्छु, जसले आफ्नै सूचना केन्द्र विकास गरिरहेको छ। म प्रायः जाभामा वसन्त बुट प्रयोग गरेर विकास गर्छु र परियोजनामा ​​उत्पन्न हुने विभिन्न प्राविधिक समस्याहरू समाधान गर्छु।

हाम्रा धेरैजसो माइक्रोसेवाहरूले सन्देश ब्रोकर मार्फत एकअर्कासँग एसिन्क्रोनस रूपमा सञ्चार गर्छन्। पहिले, हामीले IBM MQ लाई ब्रोकरको रूपमा प्रयोग गर्यौं, जसले अब लोडसँग सामना गर्न सक्दैन, तर एकै समयमा उच्च डेलिभरी ग्यारेन्टीहरू थिए।

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

ग्यारेन्टी वितरण र अधिक

तल छलफल गरिएका सेटिङहरूले पूर्वनिर्धारित जडान सेटिङहरूसँग धेरै समस्याहरू रोक्न मद्दत गर्नेछ। तर पहिले म एक प्यारामिटरमा ध्यान दिन चाहन्छु जसले सम्भावित डिबगलाई सुविधा दिन्छ।

यसले मद्दत गर्नेछ client.id उत्पादक र उपभोक्ताको लागि। पहिलो नजरमा, तपाईले अनुप्रयोगको नामलाई मानको रूपमा प्रयोग गर्न सक्नुहुन्छ, र धेरै जसो अवस्थामा यसले काम गर्नेछ। यद्यपि स्थिति जब एप्लिकेसनले धेरै उपभोक्ताहरू प्रयोग गर्दछ र तपाईंले तिनीहरूलाई उही client.id दिनुहुन्छ, निम्न चेतावनीको परिणाम:

org.apache.kafka.common.utils.AppInfoParser — Error registering AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0

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

अब ग्यारेन्टी सन्देश डेलिभरीको लागि परिदृश्य हेरौं। काफ्का निर्मातासँग एक प्यारामिटर छ acks, जसले तपाईंलाई क्लस्टर नेताले सफलतापूर्वक लेखिएको सन्देशलाई विचार गर्न आवश्यक कति स्वीकार गरेपछि कन्फिगर गर्न अनुमति दिन्छ। यो प्यारामिटरले निम्न मान लिन सक्छ:

• ० - मान्यतालाई विचार गरिने छैन।
• 1 पूर्वनिर्धारित प्यारामिटर हो, केवल 1 प्रतिकृति स्वीकार गर्न आवश्यक छ।
• −1 — सबै सिङ्क्रोनाइज्ड प्रतिकृतिहरूबाट स्वीकार गर्न आवश्यक छ (क्लस्टर सेटअप min.insync.replicas).

सूचीबद्ध मानहरूबाट यो स्पष्ट छ कि acks बराबर −1 ले सन्देश हराउने छैन भन्ने बलियो ग्यारेन्टी दिन्छ।

हामी सबैलाई थाहा छ, वितरण प्रणालीहरू अविश्वसनीय छन्। क्षणिक गल्तीहरू विरुद्ध सुरक्षा गर्न, काफ्का निर्माताले विकल्प प्रदान गर्दछ पुन: प्रयास गर्दछ, जसले तपाईंलाई भित्र पुन: पठाउने प्रयासहरूको संख्या सेट गर्न अनुमति दिन्छ डेलिभरी.टाइमआउट.एमएस। पुन: प्रयास प्यारामिटरमा Integer.MAX_VALUE (2147483647) को पूर्वनिर्धारित मान भएकोले, सन्देश पुन: प्रयासहरूको संख्या केवल delivery.timeout.ms परिवर्तन गरेर समायोजन गर्न सकिन्छ।

हामी ठ्याक्कै एक पटक डेलिभरी तिर लागिरहेका छौं

सूचीबद्ध सेटिङहरूले हाम्रो निर्मातालाई उच्च ग्यारेन्टीसहित सन्देशहरू पठाउन अनुमति दिन्छ। अब काफ्का विषयमा एउटा सन्देशको एउटा प्रतिलिपि मात्रै लेखिएको छ भनी कसरी सुनिश्चित गर्ने भन्ने कुरा गरौं? सरल अवस्थामा, यो गर्नको लागि, तपाईंले निर्माता मा प्यारामिटर सेट गर्न आवश्यक छ सक्षम। idempotence सत्य को लागी। इडम्पोटेन्सीले ग्यारेन्टी दिन्छ कि एउटा विषयको विशेष विभाजनमा एउटा सन्देश मात्र लेखिएको छ। इडम्पोटेन्सी सक्षम गर्नको लागि पूर्व शर्त मानहरू हो acks = all, पुन: प्रयास > 0, max.in.flight.requests.per.connection ≤ ५। यदि यी प्यारामिटरहरू विकासकर्ताद्वारा निर्दिष्ट गरिएको छैन भने, माथिका मानहरू स्वचालित रूपमा सेट हुनेछन्।

जब idempotency कन्फिगर गरिएको छ, यो सुनिश्चित गर्न आवश्यक छ कि समान सन्देशहरू प्रत्येक पटक समान विभाजनहरूमा समाप्त हुन्छ। यो partitioner.class कुञ्जी र प्यारामिटरलाई निर्मातामा सेट गरेर गर्न सकिन्छ। कुञ्जीबाट सुरु गरौं। यो प्रत्येक सबमिशनको लागि समान हुनुपर्छ। यो सजिलैसँग मूल पोस्टबाट कुनै पनि व्यवसाय आईडीहरू प्रयोग गरेर प्राप्त गर्न सकिन्छ। partitioner.class प्यारामिटरमा पूर्वनिर्धारित मान छ - पूर्वनिर्धारित विभाजनकर्ता। यस विभाजन रणनीतिको साथ, पूर्वनिर्धारित रूपमा हामी यसरी काम गर्छौं:

• यदि सन्देश पठाउँदा विभाजन स्पष्ट रूपमा निर्दिष्ट गरिएको छ भने, हामी यसलाई प्रयोग गर्छौं।
• यदि विभाजन निर्दिष्ट गरिएको छैन, तर कुञ्जी निर्दिष्ट गरिएको छ भने, कुञ्जीको ह्यासद्वारा विभाजन चयन गर्नुहोस्।
• यदि विभाजन र कुञ्जी निर्दिष्ट गरिएको छैन भने, विभाजनहरू एक एक गरी चयन गर्नुहोस् (राउण्ड-रोबिन)।

साथै, प्यारामिटरको साथ कुञ्जी र idempotent पठाउने प्रयोग गर्दै max.in.flight.requests.per.connection = 1 तपाईंलाई उपभोक्तामा सुव्यवस्थित सन्देश प्रशोधन दिन्छ। यो पनि याद गर्न लायक छ कि यदि पहुँच नियन्त्रण तपाइँको क्लस्टरमा कन्फिगर गरिएको छ भने, तपाइँलाई कुनै विषयमा इमानदारीपूर्वक लेख्न अधिकार चाहिन्छ।

यदि तपाईमा अचानक कुञ्जीद्वारा पठाउने क्षमताको कमी छ वा निर्माता पक्षमा तर्कले विभिन्न विभाजनहरू बीच डेटा स्थिरता कायम राख्न आवश्यक छ भने, तब लेनदेनहरू उद्धारमा आउनेछन्। थप रूपमा, चेन लेनदेन प्रयोग गरेर, तपाइँ सशर्त रूपमा काफ्कामा रेकर्ड सिङ्क्रोनाइज गर्न सक्नुहुन्छ, उदाहरणका लागि, डाटाबेसमा रेकर्डको साथ। निर्मातालाई लेनदेन पठाउन सक्षम गर्न, यो निर्दोष र थप सेट हुनुपर्छ transactional.id। यदि तपाइँको काफ्का क्लस्टरमा पहुँच नियन्त्रण कन्फिगर गरिएको छ भने, एक idempotent रेकर्ड जस्तै लेनदेन रेकर्डलाई लेखन अनुमति चाहिन्छ, जुन transactional.id मा भण्डार गरिएको मान प्रयोग गरेर मास्कद्वारा प्रदान गर्न सकिन्छ।

औपचारिक रूपमा, कुनै पनि स्ट्रिङ, जस्तै अनुप्रयोग नाम, लेनदेन पहिचानकर्ताको रूपमा प्रयोग गर्न सकिन्छ। तर यदि तपाईंले एउटै अनुप्रयोगको धेरै उदाहरणहरू एउटै transactional.id मार्फत सुरु गर्नुभयो भने, त्यसपछि पहिलो लन्च गरिएको उदाहरण त्रुटिको साथ रोकिनेछ, किनकि काफ्काले यसलाई एक जोम्बी प्रक्रिया मान्नुहुनेछ।

org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.

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

निर्माता कन्फिगर गरिएको छ, तर काफ्कामा हुने लेनदेनले सन्देशको दायरा मात्र नियन्त्रण गर्छ। लेनदेन स्थिति जेसुकै भए पनि, सन्देश तुरुन्तै शीर्षकमा जान्छ, तर अतिरिक्त प्रणाली विशेषताहरू छन्।

त्यस्ता सन्देशहरूलाई उपभोक्ताले समय अगावै पढ्नबाट रोक्नको लागि, उसले प्यारामिटर सेट गर्न आवश्यक छ isolation.level पढ्न_प्रतिबद्ध मान। यस्तो उपभोक्ताले पहिले जस्तै गैर-लेनदेन सन्देशहरू पढ्न सक्षम हुनेछन्, र लेनदेन सन्देशहरू एक प्रतिबद्धता पछि मात्र।
यदि तपाईंले पहिले सूचीबद्ध गरिएका सबै सेटिङहरू सेट गर्नुभएको छ भने, तपाईंले एक पटक डेलिभरी ठीक कन्फिगर गर्नुभएको छ। बधाई छ!

तर त्यहाँ एक थप nuance छ। Transactional.id, जुन हामीले माथि कन्फिगर गरेका छौं, वास्तवमा लेनदेन उपसर्ग हो। लेनदेन प्रबन्धकमा, एक अनुक्रम नम्बर यसमा थपिएको छ। प्राप्त पहिचानकर्ता जारी गरिएको छ transactional.id.expiration.ms, जुन काफ्का क्लस्टरमा कन्फिगर गरिएको छ र "7 दिन" को पूर्वनिर्धारित मान छ। यदि यस अवधिमा एप्लिकेसनले कुनै सन्देशहरू प्राप्त गरेको छैन भने, त्यसपछि तपाईंले अर्को लेनदेन पठाउने प्रयास गर्दा तपाईंले प्राप्त गर्नुहुनेछ अमान्यPidMappingException। लेनदेन समन्वयकले त्यसपछि अर्को लेनदेनको लागि नयाँ अनुक्रम नम्बर जारी गर्नेछ। यद्यपि, सन्देश हराउन सक्छ यदि InvalidPidMappingException सही रूपमा ह्यान्डल गरिएको छैन।

नतिजाको सट्टा

तपाईले देख्न सक्नुहुन्छ, काफ्कालाई सन्देशहरू पठाउनु मात्र पर्याप्त छैन। तपाईंले प्यारामिटरहरूको संयोजन छनौट गर्न आवश्यक छ र द्रुत परिवर्तनहरू गर्न तयार हुनुहोस्। यस लेखमा, मैले ठ्याक्कै एक पटक डेलिभरी सेटअपको विवरण देखाउने प्रयास गरें र हामीले सामना गरेका client.id र transactional.id कन्फिगरेसनहरूसँग धेरै समस्याहरू वर्णन गरें। तल निर्माता र उपभोक्ता सेटिङहरूको सारांश छ।

निर्माता:

1. acks = सबै
2. पुन: प्रयास गर्नुहोस् > ०
3. enable.idempotence = सत्य
4. max.in.flight.requests.per.connection ≤ 5 (1 व्यवस्थित पठाउनको लागि)
5. transactional.id = ${application-name}-${hostname}

उपभोक्ता:

1. isolation.level = read_committed

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

यहाँ आत्म-अध्ययनका लागि केही सामग्रीहरू छन्:

• KIP-98 - ठ्याक्कै एक पटक डेलिभरी र लेनदेन सन्देश
• सेटिङहरूको विवरण

स्रोत: www.habr.com