काफ्का वास्तव कसा बनला

अहो हाब्र!

मी टिंकॉफ टीमवर काम करतो, जे स्वतःचे सूचना केंद्र विकसित करत आहे. मी जावामध्ये स्प्रिंग बूट वापरून विकसित करतो आणि प्रोजेक्टमध्ये उद्भवणाऱ्या विविध तांत्रिक समस्या सोडवतो.

आमच्या बहुतेक मायक्रोसर्व्हिसेस मेसेज ब्रोकरद्वारे एकमेकांशी असिंक्रोनसपणे संवाद साधतात. पूर्वी, आम्ही ब्रोकर म्हणून 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 मूल्य म्हणून विषयाचे नाव संयोजन वापरणे चांगले. आमच्या कॉन्फिगरेशनचा परिणाम कमांड आउटपुटमध्ये दिसू शकतो kafka-ग्राहक-गट Confluent कडून युटिलिटीज कडून:

आता गॅरंटीड मेसेज डिलिव्हरीसाठी परिस्थिती पाहू. काफ्का निर्मात्याकडे एक पॅरामीटर आहे acks, जे तुम्हाला क्लस्टर लीडरला यशस्वीरित्या लिहिलेल्या संदेशाचा विचार करण्यासाठी किती मान्यतेनंतर कॉन्फिगर करण्याची परवानगी देते. हे पॅरामीटर खालील मूल्ये घेऊ शकते:

• 0 - पावती विचारात घेतली जाणार नाही.
• 1 हे डीफॉल्ट पॅरामीटर आहे, फक्त 1 प्रतिकृती मान्य करण्यासाठी आवश्यक आहे.
• −1 — सर्व सिंक्रोनाइझ केलेल्या प्रतिकृतींकडून पावती आवश्यक आहे (क्लस्टर सेटअप min.insync.replicas).

सूचीबद्ध मूल्यांवरून हे स्पष्ट आहे की −1 च्या बरोबरीचे acks संदेश गमावले जाणार नाही याची सर्वात मजबूत हमी देते.

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

आम्ही नक्की एकदाच डिलिव्हरीच्या दिशेने वाटचाल करत आहोत

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

जेव्हा idempotency कॉन्फिगर केले जाते, तेव्हा प्रत्येक वेळी समान संदेश समान विभाजनांमध्ये संपतील याची खात्री करणे आवश्यक आहे. हे partitioner.class की आणि पॅरामीटर प्रोड्युसरवर सेट करून केले जाऊ शकते. चला की सह प्रारंभ करूया. प्रत्येक सबमिशनसाठी ते समान असणे आवश्यक आहे. मूळ पोस्टमधील कोणताही व्यवसाय आयडी वापरून हे सहज साध्य करता येते. partitioner.class पॅरामीटरमध्ये डीफॉल्ट मूल्य असते − डीफॉल्ट विभाजनकर्ता. या विभाजन धोरणासह, डीफॉल्टनुसार आम्ही असे कार्य करतो:

• संदेश पाठवताना विभाजन स्पष्टपणे निर्दिष्ट केले असल्यास, आम्ही ते वापरतो.
• विभाजन निर्दिष्ट केले नसल्यास, परंतु की निर्दिष्ट केली असल्यास, कीच्या हॅशद्वारे विभाजन निवडा.
• जर विभाजन आणि की निर्दिष्ट केली नसेल, तर विभाजने एक-एक करून निवडा (राउंड-रॉबिन).

तसेच, पॅरामीटरसह की आणि इडम्पोटेंट पाठवणे वापरणे max.in.flight.requests.per.connection = 1 तुम्हाला ग्राहकांवर सुव्यवस्थित संदेश प्रक्रिया देते. हे देखील लक्षात ठेवण्यासारखे आहे की जर तुमच्या क्लस्टरवर ऍक्सेस कंट्रोल कॉन्फिगर केले असेल, तर तुम्हाला एखाद्या विषयावर विचारपूर्वक लिहिण्यासाठी अधिकारांची आवश्यकता असेल.

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

पण आणखी एक बारकावे आहे. Transactional.id, जो आम्ही वर कॉन्फिगर केला आहे, हा प्रत्यक्षात व्यवहार उपसर्ग आहे. ट्रान्झॅक्शन मॅनेजरवर, त्यात एक क्रम क्रमांक जोडला जातो. प्राप्त आयडेंटिफायर यांना जारी केले जाते transactional.id.expiration.ms, जे काफ्का क्लस्टरवर कॉन्फिगर केले आहे आणि "7 दिवस" ​​चे डीफॉल्ट मूल्य आहे. जर या कालावधीत अनुप्रयोगास कोणतेही संदेश प्राप्त झाले नाहीत, तर जेव्हा आपण पुढील व्यवहार करण्याचा प्रयत्न कराल तेव्हा आपल्याला संदेश प्राप्त होईल अवैधPidMappingException. त्यानंतर व्यवहार समन्वयक पुढील व्यवहारासाठी नवीन अनुक्रम क्रमांक जारी करेल. तथापि, InvalidPidMappingException योग्यरित्या हाताळले नसल्यास संदेश गमावला जाऊ शकतो.

निकालाऐवजी

तुम्ही बघू शकता, काफ्काला फक्त संदेश पाठवणे पुरेसे नाही. तुम्हाला पॅरामीटर्सचे संयोजन निवडण्याची आणि जलद बदल करण्यासाठी तयार राहण्याची आवश्यकता आहे. या लेखात, मी डिलिव्हरी सेटअप एकदाच तपशीलवार दाखवण्याचा प्रयत्न केला आणि आम्हाला आलेल्या client.id आणि transactional.id कॉन्फिगरेशनमधील अनेक समस्यांचे वर्णन केले. खाली निर्माता आणि ग्राहक सेटिंग्जचा सारांश आहे.

निर्माते:

1. acks = सर्व
2. पुन्हा प्रयत्न करा > 0
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