კაფკასგან მიღებული მოვლენების გადამუშავება

ჰეი ჰაბრ.

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

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

როგორ იყო კაფკამდე

ადრე პროექტში ჩვენ ვიყენებდით IBM MQ შეტყობინებების ასინქრონული მიწოდებისთვის. თუ რაიმე შეცდომა მოხდა სერვისის მუშაობის დროს, მიღებული შეტყობინება შეიძლება განთავსდეს მკვდარი ასოების რიგში (DLQ) შემდგომი ხელით ანალიზისათვის. DLQ შეიქმნა შემომავალი რიგის გვერდით, შეტყობინება გადავიდა IBM MQ-ში.

თუ შეცდომა დროებითი იყო და შეგვეძლო მისი დადგენა (მაგალითად, ResourceAccessException HTTP ზარზე ან MongoTimeoutException MongoDb მოთხოვნაზე), მაშინ ხელახლა ცდის სტრატეგია ძალაში შევა. აპლიკაციის განშტოების ლოგიკის მიუხედავად, თავდაპირველი შეტყობინება გადავიდა ან სისტემის რიგში დაგვიანებული გაგზავნისთვის, ან ცალკეულ აპლიკაციაში, რომელიც დიდი ხნის წინ გაკეთდა შეტყობინებების ხელახლა გასაგზავნად. ეს მოიცავს ხელახლა გაგზავნის ნომერს შეტყობინების სათაურში, რომელიც დაკავშირებულია დაყოვნების ინტერვალთან ან აპლიკაციის დონის სტრატეგიის დასასრულთან. თუ ჩვენ მივაღწიეთ სტრატეგიის დასასრულს, მაგრამ გარე სისტემა ჯერ კიდევ მიუწვდომელია, მაშინ შეტყობინება განთავსდება DLQ-ში ხელით გაანალიზებისთვის.

გადაწყვეტის ძიება

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

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

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

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

Spring-kafka თავად უზრუნველყოფს ContainerAwareErrorHandler-ის რამდენიმე განხორციელებას, მაგალითად SeekToCurrentError Handler, რომლითაც შეცდომის შემთხვევაში შეცდომის შემთხვევაში მოგვიანებით შეძლებთ შეტყობინების დამუშავებას. Spring-kafka 2.3 ვერსიით დაწყებული, შესაძლებელი გახდა BackOffPolicy-ის დაყენება.

ეს მიდგომა საშუალებას აძლევს ხელახლა დამუშავებულ შეტყობინებებს გადარჩეს აპლიკაციის გადატვირთვაში, მაგრამ ჯერ კიდევ არ არსებობს DLQ მექანიზმი. ჩვენ ავირჩიეთ ეს ვარიანტი 2019 წლის დასაწყისში, ოპტიმისტურად გვჯეროდა, რომ DLQ არ იქნებოდა საჭირო (გაგვიმართლა და რეალურად არ დაგვჭირდა აპლიკაციის ასეთი გადამუშავების სისტემით მუშაობის რამდენიმე თვის შემდეგ). დროებითმა შეცდომებმა გამოიწვია SeekToCurrentErrorHandler-ის გააქტიურება. დარჩენილი შეცდომები დაიბეჭდა ჟურნალში, რის შედეგადაც მოხდა ოფსეტური და დამუშავება გაგრძელდა შემდეგი შეტყობინებით.

Საბოლოო გადაწყვეტილება

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

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

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

მომხმარებლის შეჩერება ხაზოვანი ლოგიკის აპლიკაციისთვის

Spring-kafka-თან მუშაობისას, მომხმარებლის შეჩერების კოდი შეიძლება ასე გამოიყურებოდეს:

public void pauseListenerContainer(MessageListenerContainer listenerContainer, 
                                   Instant retryAt) {
        if (nonNull(retryAt) && listenerContainer.isRunning()) {
            listenerContainer.stop();
            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
            return;
        }
        // to DLQ
    }

მაგალითში, retryAt არის MessageListenerContainer-ის გადატვირთვის დრო, თუ ის ჯერ კიდევ მუშაობს. ხელახალი გაშვება მოხდება TaskScheduler-ში გაშვებულ ცალკეულ თემაში, რომლის განხორციელებაც ასევე უზრუნველყოფილია გაზაფხულზე.

ჩვენ ვპოულობთ retryAt მნიშვნელობას შემდეგი გზით:

1. ხელახალი გამოძახების მრიცხველის ღირებულება მოძებნილია.
2. მრიცხველის მნიშვნელობიდან გამომდინარე, განმეორებითი სტრატეგიის დაყოვნების მიმდინარე ინტერვალი მოძებნილია. სტრატეგია დეკლარირებულია თავად აპლიკაციაში; ჩვენ ავირჩიეთ JSON ფორმატი მის შესანახად.
3. JSON მასივში ნაპოვნი ინტერვალი შეიცავს წამების რაოდენობას, რის შემდეგაც დამუშავება უნდა განმეორდეს. წამების ეს რაოდენობა ემატება მიმდინარე დროს, რათა შეიქმნას მნიშვნელობა retryAt-ისთვის.
4. თუ ინტერვალი ვერ მოიძებნა, მაშინ retryAt-ის მნიშვნელობა არის null და შეტყობინება გაეგზავნება DLQ-ს ხელით ანალიზისათვის.

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

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

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

არადეტერმინისტული ლოგიკის მქონე აპლიკაციებიდან შეტყობინებების დამუშავების ცალკე აპლიკაცია

აქ არის კოდის მაგალითი, რომელიც აგზავნის შეტყობინებას ასეთ აპლიკაციაში (განმეორებითი მცდელობა), რომელიც ხელახლა გაიგზავნება DESTINATION თემაში, როდესაც RETRY_AT დრო მიაღწევს:

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-ის მნიშვნელობა მოიძებნება ისევე, როგორც განმეორებითი ცდის მექანიზმისთვის Consumer stop-ის მეშვეობით. DESTINATION-ისა და RETRY_AT-ის გარდა, ჩვენ გავდივართ:

• GROUP_ID, რომლითაც ვაჯგუფებთ შეტყობინებებს ხელით ანალიზისა და გამარტივებული ძიებისთვის.
• ORIGINAL_PARTITION შეეცადოთ შეინარჩუნოთ იგივე მომხმარებელი ხელახალი დამუშავებისთვის. ეს პარამეტრი შეიძლება იყოს null, ამ შემთხვევაში ახალი დანაყოფი მიიღება ორიგინალური შეტყობინების record.key() კლავიშის გამოყენებით.
• განახლებულია COUNTER მნიშვნელობა ხელახლა ცდის სტრატეგიის შესასრულებლად.
• SEND_TO არის მუდმივი, რომელიც მიუთითებს, იგზავნება თუ არა შეტყობინება ხელახალი დამუშავებისთვის RETRY_AT-ზე მიღწევის შემდეგ, თუ განთავსებულია DLQ-ში.
• REASON - შეტყობინებების დამუშავების შეწყვეტის მიზეზი.

Retryer ინახავს შეტყობინებებს ხელახლა გასაგზავნად და ხელით გაანალიზებისთვის PostgreSQL-ში. ტაიმერი იწყებს დავალებას, რომელიც პოულობს შეტყობინებებს RETRY_AT-ით და აბრუნებს მათ DESTINATION თემის ORIGINAL_PARTITION განყოფილებაში გასაღებით record.key().

გაგზავნის შემდეგ, შეტყობინებები წაიშლება PostgreSQL-დან. შეტყობინებების ხელით ანალიზი ხდება მარტივ ინტერფეისში, რომელიც ურთიერთქმედებს Retryer-თან REST API-ის მეშვეობით. მისი ძირითადი მახასიათებლებია DLQ-დან შეტყობინებების ხელახლა გაგზავნა ან წაშლა, შეცდომის ინფორმაციის ნახვა და შეტყობინებების ძიება, მაგალითად, შეცდომის სახელით.

ვინაიდან წვდომის კონტროლი ჩართულია ჩვენს კლასტერებზე, აუცილებელია დამატებით მოითხოვოთ წვდომა იმ თემაზე, რომელსაც Retryer უსმენს და მიეცით საშუალება Retryer-ს დაწეროს DESTINATION თემაში. ეს არასასიამოვნოა, მაგრამ, ინტერვალის თემის მიდგომისგან განსხვავებით, ჩვენ გვაქვს სრულფასოვანი DLQ და UI მის სამართავად.

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

ნაგულისხმევად, ეს მიდგომა არ უზრუნველყოფს ამომრთველის ფუნქციონირებას, თუმცა ის შეიძლება დაემატოს აპლიკაციას გამოყენებით Spring-cloud-Netflix ან ახალი გაზაფხულის ღრუბლის ამომრთველი, ადგილების შეფუთვა, სადაც გარე სერვისები მოწოდებულია შესაბამის აბსტრაქციებში. გარდა ამისა, შესაძლებელი ხდება სტრატეგიის არჩევა შუალედი ნიმუში, რომელიც ასევე შეიძლება სასარგებლო იყოს. მაგალითად, Spring-cloud-netflix-ში ეს შეიძლება იყოს ძაფის აუზი ან სემაფორი.

გამოყვანის

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

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

წყარო: www.habr.com