როგორ გახდა კაფკა რეალობა

ჰეი ჰაბრ!

მე ვმუშაობ Tinkoff-ის გუნდზე, რომელიც ავითარებს საკუთარ შეტყობინებების ცენტრს. მე ძირითადად Java-ში ვვითარდები Spring boot-ის გამოყენებით და ვხსნი სხვადასხვა ტექნიკურ პრობლემებს, რომლებიც წარმოიქმნება პროექტში.

ჩვენი მიკროსერვისების უმეტესობა ერთმანეთთან ასინქრონულად ურთიერთობს შეტყობინებების ბროკერის საშუალებით. მანამდე ბროკერად ვიყენებდით IBM MQ-ს, რომელიც დატვირთვას ვეღარ უმკლავდებოდა, ამავდროულად ჰქონდა მიწოდების მაღალი გარანტიები.

შემცვლელად შემოგვთავაზეს Apache Kafka, რომელსაც აქვს მაღალი სკალირების პოტენციალი, მაგრამ, სამწუხაროდ, მოითხოვს კონფიგურაციის თითქმის ინდივიდუალურ მიდგომას სხვადასხვა სცენარისთვის. გარდა ამისა, ერთხელ მაინც მიწოდების მექანიზმი, რომელიც მუშაობს კაფკაში ნაგულისხმევად, არ იძლეოდა კონსისტენციის საჭირო დონის შენარჩუნებას. შემდეგი, მე გაგიზიარებთ ჩვენს გამოცდილებას კაფკას კონფიგურაციაში, კერძოდ, გეტყვით, თუ როგორ უნდა დააკონფიგურიროთ და იცხოვროთ ზუსტად ერთხელ მიწოდებით.

გარანტირებული მიწოდება და სხვა

ქვემოთ განხილული პარამეტრები დაგეხმარებათ თავიდან აიცილოთ მრავალი პრობლემა ნაგულისხმევი კავშირის პარამეტრებთან დაკავშირებით. მაგრამ პირველ რიგში მინდა მივაქციო ყურადღება ერთ პარამეტრს, რომელიც ხელს შეუწყობს შესაძლო გამართვას.

ეს დაეხმარება კლიენტი.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 მნიშვნელობა. ჩვენი კონფიგურაციის შედეგი ჩანს ბრძანების გამომავალში კაფკა-მომხმარებელთა ჯგუფები კომუნალური პროგრამებიდან Confluent-დან:

ახლა მოდით შევხედოთ სცენარს გარანტირებული შეტყობინების მიწოდებისთვის. კაფკას პროდიუსერს აქვს პარამეტრი კისრები, რომელიც საშუალებას გაძლევთ დააკონფიგურიროთ რამდენი აღიარების შემდეგ სჭირდება კლასტერის ლიდერს, რათა განიხილოს შეტყობინება წარმატებით დაწერილი. ამ პარამეტრს შეუძლია მიიღოს შემდეგი მნიშვნელობები:

• 0 — აღიარება არ განიხილება.
• 1 არის ნაგულისხმევი პარამეტრი, მხოლოდ 1 ასლის დადასტურებაა საჭირო.
• −1 - საჭიროა ყველა სინქრონიზებული რეპლიკიდან აღიარება (კლასტერული დაყენება min.insync.replicas).

ჩამოთვლილი მნიშვნელობებიდან ირკვევა, რომ −1-ის ტოლი acks იძლევა უძლიერეს გარანტიას, რომ შეტყობინება არ დაიკარგება.

როგორც ყველამ ვიცით, განაწილებული სისტემები არასანდოა. გარდამავალი ხარვეზებისგან თავის დასაცავად, Kafka Producer გთავაზობთ ვარიანტს ხელახლა ცდილობს, რომელიც საშუალებას გაძლევთ დააყენოთ ხელახლა გაგზავნის მცდელობების რაოდენობა შიგნით მიწოდება.timeout.ms. ვინაიდან განმეორებითი ცდების პარამეტრს აქვს ნაგულისხმევი მნიშვნელობა Integer.MAX_VALUE (2147483647), შეტყობინების ხელახალი ცდების რაოდენობის კორექტირება შესაძლებელია მხოლოდ delivery.timeout.ms შეცვლით.

ჩვენ მივდივართ ზუსტად ერთხელ მიწოდებისკენ

ჩამოთვლილი პარამეტრები საშუალებას აძლევს ჩვენს პროდიუსერს მიაწოდოს შეტყობინებები მაღალი გარანტიით. მოდით ვისაუბროთ იმაზე, თუ როგორ უნდა დავრწმუნდეთ, რომ შეტყობინების მხოლოდ ერთი ასლი დაიწერება კაფკას თემაზე? უმარტივეს შემთხვევაში, ამისათვის თქვენ უნდა დააყენოთ პარამეტრი Producer-ზე ჩართვა.იდეპოტენცია ჭეშმარიტად. Idempotency გარანტიას იძლევა, რომ მხოლოდ ერთი შეტყობინება იწერება ერთი თემის კონკრეტულ დანაყოფზე. იმპოტენციის გააქტიურების წინაპირობა ღირებულებებია acks = ყველა, ხელახლა სცადეთ > 0, max.in.flight.requests.per.connection ≤ 5. თუ ეს პარამეტრები არ არის მითითებული დეველოპერის მიერ, ზემოთ ჩამოთვლილი მნიშვნელობები ავტომატურად დაყენდება.

როდესაც idempotency არის კონფიგურირებული, აუცილებელია იმის უზრუნველყოფა, რომ ერთი და იგივე შეტყობინებები ყოველ ჯერზე ერთსა და იმავე დანაყოფებში აღმოჩნდეს. ეს შეიძლება გაკეთდეს partitioner.class გასაღების და პარამეტრის Producer-ზე დაყენებით. დავიწყოთ გასაღებით. ეს უნდა იყოს იგივე თითოეული წარდგენისთვის. ამის მარტივად მიღწევა შესაძლებელია ორიგინალური პოსტიდან ნებისმიერი ბიზნეს ID-ის გამოყენებით. partitioner.class პარამეტრს აქვს ნაგულისხმევი მნიშვნელობა − ნაგულისხმევი გამყოფი. ამ დაყოფის სტრატეგიით, ნაგულისხმევად ჩვენ ასე ვიმოქმედებთ:

• თუ დანაყოფი პირდაპირ არის მითითებული შეტყობინების გაგზავნისას, მაშინ ჩვენ ვიყენებთ მას.
• თუ დანაყოფი არ არის მითითებული, მაგრამ გასაღები მითითებულია, აირჩიეთ დანაყოფი გასაღების ჰეშის მიხედვით.
• თუ დანაყოფი და გასაღები არ არის მითითებული, შეარჩიეთ ტიხრები სათითაოდ (მრგვალი რგოლი).

ასევე, გასაღების გამოყენება და პარამეტრით იდემპოტენტური გაგზავნა max.in.flight.requests.per.connection = 1 გაძლევთ მომხმარებელზე შეტყობინებების გამარტივებულ დამუშავებას. ასევე უნდა გვახსოვდეს, რომ თუ წვდომის კონტროლი კონფიგურირებულია თქვენს კლასტერზე, მაშინ დაგჭირდებათ უფლება, რომ უაზროდ ჩაწეროთ თემა.

თუ მოულოდნელად არ გაქვთ გასაღებით იდემპოტენტური გაგზავნის შესაძლებლობები ან პროდიუსერის მხრიდან ლოგიკა მოითხოვს მონაცემთა თანმიმდევრულობის შენარჩუნებას სხვადასხვა დანაყოფებს შორის, მაშინ ტრანზაქციები გადაარჩენს. გარდა ამისა, ჯაჭვური ტრანზაქციის გამოყენებით, შეგიძლიათ პირობითად მოახდინოთ ჩანაწერის სინქრონიზაცია კაფკაში, მაგალითად, მონაცემთა ბაზაში ჩანაწერთან. პროდიუსერთან ტრანზაქციის გაგზავნის გასააქტიურებლად, ის უნდა იყოს არაპოტენტური და დამატებით დაყენებული ტრანზაქციული.id. თუ წვდომის კონტროლი კონფიგურირებულია თქვენს Kafka კლასტერზე, მაშინ ტრანზაქციის ჩანაწერს, ისევე როგორც იდემპოტენტურ ჩანაწერს, დასჭირდება ჩაწერის უფლებები, რაც შეიძლება მინიჭებული იყოს ნიღბით ტრანსაქციაში შენახული მნიშვნელობის გამოყენებით.

ფორმალურად, ნებისმიერი სტრიქონი, როგორიცაა განაცხადის სახელი, შეიძლება გამოყენებულ იქნას როგორც ტრანზაქციის იდენტიფიკატორი. მაგრამ თუ თქვენ გაუშვით ერთი და იგივე აპლიკაციის რამდენიმე ინსტანცია ერთი და იგივე transaksional.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.

ამ პრობლემის გადასაჭრელად განაცხადის სახელს ვამატებთ სუფიქსს ჰოსტის სახელის სახით, რომელსაც ვიღებთ გარემოს ცვლადებიდან.

პროდიუსერი კონფიგურირებულია, მაგრამ კაფკაზე ტრანზაქციები აკონტროლებს მხოლოდ შეტყობინების ფარგლებს. ტრანზაქციის სტატუსის მიუხედავად, შეტყობინება დაუყოვნებლივ გადადის თემაში, მაგრამ აქვს დამატებითი სისტემის ატრიბუტები.

იმისათვის, რომ მომხმარებელმა არ წაიკითხოს ასეთი შეტყობინებები ვადაზე ადრე, მან უნდა დააყენოს პარამეტრი იზოლაცია.დონე read_committed მნიშვნელობისთვის. ასეთ მომხმარებელს შეეძლება წაიკითხოს არატრანზაქციის შეტყობინებები, როგორც ადრე, ხოლო ტრანზაქციის შეტყობინებები მხოლოდ ჩადენის შემდეგ.
თუ დაყენებული გაქვთ ადრე ჩამოთვლილი ყველა პარამეტრი, მაშინ თქვენ დააკონფიგურირეთ ზუსტად ერთხელ მიწოდება. გილოცავ!

მაგრამ არის კიდევ ერთი ნიუანსი. Transactional.id, რომელიც ჩვენ ზემოთ დავაყენეთ, რეალურად არის ტრანზაქციის პრეფიქსი. ტრანზაქციის მენეჯერზე მას ემატება რიგითი ნომერი. მიღებული იდენტიფიკატორი გაიცემა transaksional.id.expiration.ms, რომელიც კონფიგურირებულია კაფკას კლასტერზე და აქვს ნაგულისხმევი მნიშვნელობა "7 დღე". თუ ამ დროის განმავლობაში აპლიკაციას არ მიუღია შეტყობინებები, მაშინ, როდესაც ცდილობთ შემდეგი ტრანზაქციის გაგზავნას, მიიღებთ InvalidPidMappingException. შემდეგ ტრანზაქციის კოორდინატორი გასცემს ახალ რიგით ნომერს შემდეგი ტრანზაქციისთვის. თუმცა, შეტყობინება შეიძლება დაიკარგოს, თუ InvalidPidMappingException არ არის სწორად დამუშავებული.

შედეგების ნაცვლად

როგორც ხედავთ, კაფკას უბრალოდ მესიჯების გაგზავნა საკმარისი არ არის. თქვენ უნდა აირჩიოთ პარამეტრების კომბინაცია და მზად იყოთ სწრაფი ცვლილებებისთვის. ამ სტატიაში შევეცადე დეტალურად მეჩვენებინა ზუსტად ერთხელ მიწოდების დაყენება და აღვწერე რამდენიმე პრობლემა client.id და transaksional.id კონფიგურაციებთან, რომლებიც შეგვხვდა. ქვემოთ მოცემულია მწარმოებლისა და მომხმარებლის პარამეტრების შეჯამება.

პროდიუსერი:

1. acks = ყველა
2. განმეორებითი ცდები > 0
3. ჩართვა.იდეპოტენცია = ჭეშმარიტი
4. max.in.flight.requests.per.connection ≤ 5 (1 მოწესრიგებული გაგზავნისთვის)
5. transaksional.id = ${aplication-name}-${hostname}

მომხმარებელი:

1. იზოლაცია.დონე = წაკითხული_ჩამოწერილი

სამომავლო აპლიკაციებში შეცდომების შესამცირებლად, ჩვენ გავაკეთეთ საკუთარი შეფუთვა ზამბარის კონფიგურაციაზე, სადაც უკვე დაყენებულია ზოგიერთი ჩამოთვლილი პარამეტრის მნიშვნელობები.

აქ არის რამდენიმე მასალა თვითშესწავლისთვის:

• KIP-98 - ზუსტად ერთხელ მიწოდება და ტრანზაქციის შეტყობინებები
• პარამეტრების აღწერა

წყარო: www.habr.com