Πώς ο Κάφκα έγινε πραγματικότητα

Γεια σου Χαμπρ!

Εργάζομαι στην ομάδα Tinkoff, η οποία αναπτύσσει το δικό της κέντρο ειδοποιήσεων. Αναπτύσσομαι κυρίως σε Java χρησιμοποιώντας Spring boot και λύνω διάφορα τεχνικά προβλήματα που προκύπτουν σε ένα έργο.

Οι περισσότερες από τις μικροϋπηρεσίες μας επικοινωνούν μεταξύ τους ασύγχρονα μέσω ενός μεσίτη μηνυμάτων. Παλαιότερα, χρησιμοποιούσαμε ως μεσίτη την IBM MQ, η οποία δεν μπορούσε πλέον να αντεπεξέλθει στο φορτίο, αλλά ταυτόχρονα είχε υψηλές εγγυήσεις παράδοσης.

Ως αντικατάσταση, μας προσφέρθηκε το Apache Kafka, το οποίο έχει υψηλή δυνατότητα κλιμάκωσης, αλλά, δυστυχώς, απαιτεί μια σχεδόν ατομική προσέγγιση στη διαμόρφωση για διαφορετικά σενάρια. Επιπλέον, ο μηχανισμός παράδοσης τουλάχιστον μία φορά που λειτουργεί στο Kafka από προεπιλογή δεν επέτρεπε τη διατήρηση του απαιτούμενου επιπέδου συνέπειας εκτός συσκευασίας. Στη συνέχεια, θα μοιραστώ την εμπειρία μας στη διαμόρφωση του Kafka, συγκεκριμένα, θα σας πω πώς να διαμορφώσετε και να ζήσετε με ακριβώς μία φορά παράδοση.

Εγγυημένη παράδοση και πολλά άλλα

Οι ρυθμίσεις που αναφέρονται παρακάτω θα βοηθήσουν στην αποφυγή ορισμένων προβλημάτων με τις προεπιλεγμένες ρυθμίσεις σύνδεσης. Αλλά πρώτα θα ήθελα να δώσω προσοχή σε μια παράμετρο που θα διευκολύνει έναν πιθανό εντοπισμό σφαλμάτων.

Αυτό θα βοηθήσει ταυτότητα πελάτη για Παραγωγό και Καταναλωτή. Με την πρώτη ματιά, μπορείτε να χρησιμοποιήσετε το όνομα της εφαρμογής ως τιμή και στις περισσότερες περιπτώσεις αυτό θα λειτουργήσει. Αν και η κατάσταση όταν μια εφαρμογή χρησιμοποιεί πολλούς καταναλωτές και τους δίνετε το ίδιο 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 σε μια εφαρμογή με τον Kafka, τότε αυτό μπορεί να είναι πρόβλημα. Για αυτήν την περίπτωση, είναι καλύτερο να χρησιμοποιήσετε έναν συνδυασμό του ονόματος της εφαρμογής και, για παράδειγμα, του ονόματος θέματος ως τιμή client.id. Το αποτέλεσμα της διαμόρφωσής μας φαίνεται στην έξοδο της εντολής kafka-καταναλωτικές ομάδες από βοηθητικά προγράμματα από το Confluent:

Τώρα ας δούμε το σενάριο για την εγγυημένη παράδοση μηνυμάτων. Ο Kafka Producer έχει μια παράμετρο λαβές, το οποίο σας επιτρέπει να διαμορφώσετε μετά από πόσες επιβεβαιώσεις χρειάζεται ο αρχηγός συμπλέγματος για να εξετάσει το μήνυμα επιτυχώς γραμμένο. Αυτή η παράμετρος μπορεί να λάβει τις ακόλουθες τιμές:

• 0 — αναγνώριση δεν θα ληφθεί υπόψη.
• 1 είναι η προεπιλεγμένη παράμετρος, απαιτείται μόνο 1 αντίγραφο για επιβεβαίωση.
• −1 — απαιτείται επιβεβαίωση από όλα τα συγχρονισμένα αντίγραφα (ρύθμιση συμπλέγματος min.insync.replicas).

Από τις αναφερόμενες τιμές είναι σαφές ότι τα acks ίσα με -1 παρέχουν την ισχυρότερη εγγύηση ότι το μήνυμα δεν θα χαθεί.

Όπως όλοι γνωρίζουμε, τα κατανεμημένα συστήματα είναι αναξιόπιστα. Για προστασία από παροδικά σφάλματα, η Kafka Producer παρέχει την επιλογή επαναλαμβάνει, το οποίο σας επιτρέπει να ορίσετε τον αριθμό των προσπαθειών εκ νέου αποστολής εντός delivery.timeout.ms. Εφόσον η παράμετρος επαναλήψεων έχει μια προεπιλεγμένη τιμή Integer.MAX_VALUE (2147483647), ο αριθμός των επαναλήψεων μηνυμάτων μπορεί να προσαρμοστεί αλλάζοντας μόνο delivery.timeout.ms.

Προχωράμε προς μια ακριβώς παράδοση

Οι ρυθμίσεις που αναφέρονται επιτρέπουν στον Παραγωγό μας να παραδίδει μηνύματα με υψηλή εγγύηση. Ας μιλήσουμε τώρα για το πώς να διασφαλίσουμε ότι μόνο ένα αντίγραφο ενός μηνύματος γράφεται σε ένα θέμα του Κάφκα; Στην απλούστερη περίπτωση, για να το κάνετε αυτό, πρέπει να ορίσετε την παράμετρο στο Producer ενεργοποιώ.ιδεοδυναμία στο αληθινό. Το Idempotency εγγυάται ότι μόνο ένα μήνυμα γράφεται σε ένα συγκεκριμένο διαμέρισμα ενός θέματος. Η προϋπόθεση για να ενεργοποιηθεί η ανικανότητα είναι οι αξίες acks = όλα, επανάληψη > 0, max.in.flight.requests.per.connection ≤ 5. Εάν αυτές οι παράμετροι δεν καθορίζονται από τον προγραμματιστή, οι παραπάνω τιμές θα οριστούν αυτόματα.

Όταν διαμορφώνεται το idempotency, είναι απαραίτητο να διασφαλίζετε ότι τα ίδια μηνύματα καταλήγουν στα ίδια διαμερίσματα κάθε φορά. Αυτό μπορεί να γίνει ορίζοντας το κλειδί partitioner.class και την παράμετρο σε Producer. Ας ξεκινήσουμε με το κλειδί. Πρέπει να είναι το ίδιο για κάθε υποβολή. Αυτό μπορεί εύκολα να επιτευχθεί χρησιμοποιώντας οποιοδήποτε από τα αναγνωριστικά επιχείρησης από την αρχική ανάρτηση. Η παράμετρος partitioner.class έχει μια προεπιλεγμένη τιμή − Προεπιλογή διαμερίσματος. Με αυτήν τη στρατηγική κατάτμησης, από προεπιλογή ενεργούμε ως εξής:

• Εάν το διαμέρισμα καθορίζεται ρητά κατά την αποστολή του μηνύματος, τότε το χρησιμοποιούμε.
• Εάν το διαμέρισμα δεν έχει καθοριστεί, αλλά το κλειδί έχει καθοριστεί, επιλέξτε το διαμέρισμα με τον κατακερματισμό του κλειδιού.
• Εάν το διαμέρισμα και το κλειδί δεν έχουν καθοριστεί, επιλέξτε τα διαμερίσματα ένα προς ένα (round-robin).

Επίσης, χρήση κλειδιού και ανεπαρκής αποστολή με παράμετρο max.in.flight.requests.per.connection = 1 σας παρέχει απλοποιημένη επεξεργασία μηνυμάτων στον Καταναλωτή. Αξίζει επίσης να θυμάστε ότι εάν ο έλεγχος πρόσβασης έχει ρυθμιστεί στο σύμπλεγμα σας, τότε θα χρειαστείτε δικαιώματα για να γράψετε ανίκανα σε ένα θέμα.

Εάν ξαφνικά δεν έχετε τις δυνατότητες ανεπαρκούς αποστολής με κλειδί ή η λογική από την πλευρά του Παραγωγού απαιτεί τη διατήρηση της συνέπειας των δεδομένων μεταξύ των διαφορετικών κατατμήσεων, τότε οι συναλλαγές θα βοηθήσουν. Επιπλέον, χρησιμοποιώντας μια αλυσιδωτή συναλλαγή, μπορείτε να συγχρονίσετε υπό όρους μια εγγραφή στο Kafka, για παράδειγμα, με μια εγγραφή στη βάση δεδομένων. Για να ενεργοποιηθεί η αποστολή συναλλαγών στον Παραγωγό, πρέπει να είναι ανίκανη και επιπλέον να έχει οριστεί συναλλαγής.id. Εάν το σύμπλεγμα Kafka σας έχει διαμορφώσει τον έλεγχο πρόσβασης, τότε μια εγγραφή συναλλαγής, όπως μια ανεπαρκής εγγραφή, θα χρειαστεί δικαιώματα εγγραφής, τα οποία μπορούν να εκχωρηθούν με μάσκα χρησιμοποιώντας την τιμή που είναι αποθηκευμένη στο transaksional.id.

Επίσημα, οποιαδήποτε συμβολοσειρά, όπως το όνομα της εφαρμογής, μπορεί να χρησιμοποιηθεί ως αναγνωριστικό συναλλαγής. Αλλά αν εκκινήσετε πολλές παρουσίες της ίδιας εφαρμογής με το ίδιο tradeal.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.

Για να λύσουμε αυτό το πρόβλημα, προσθέτουμε ένα επίθημα στο όνομα της εφαρμογής με τη μορφή του ονόματος κεντρικού υπολογιστή, το οποίο λαμβάνουμε από μεταβλητές περιβάλλοντος.

Ο παραγωγός έχει ρυθμιστεί, αλλά οι συναλλαγές στο Kafka ελέγχουν μόνο το εύρος του μηνύματος. Ανεξάρτητα από την κατάσταση της συναλλαγής, το μήνυμα πηγαίνει αμέσως στο θέμα, αλλά έχει πρόσθετα χαρακτηριστικά συστήματος.

Για να αποτρέψει την έγκαιρη ανάγνωση τέτοιων μηνυμάτων από τον Καταναλωτή, πρέπει να ορίσει την παράμετρο απομόνωση.επίπεδο στην τιμή read_committed. Ένας τέτοιος καταναλωτής θα μπορεί να διαβάζει μηνύματα χωρίς συναλλαγές όπως πριν, και μηνύματα συναλλαγής μόνο μετά από δέσμευση.
Εάν έχετε ορίσει όλες τις ρυθμίσεις που αναφέρονται προηγουμένως, τότε έχετε διαμορφώσει ακριβώς μία φορά την παράδοση. Συγχαρητήρια!

Υπάρχει όμως μια ακόμη απόχρωση. Το Transactional.id, το οποίο διαμορφώσαμε παραπάνω, είναι στην πραγματικότητα το πρόθεμα συναλλαγής. Στη διαχείριση συναλλαγών, προστίθεται ένας αύξων αριθμός. Το ληφθέν αναγνωριστικό εκδίδεται σε transaksional.id.expiration.ms, το οποίο έχει ρυθμιστεί σε ένα σύμπλεγμα Kafka και έχει προεπιλεγμένη τιμή "7 ημέρες". Εάν σε αυτό το διάστημα η εφαρμογή δεν έχει λάβει μηνύματα, τότε όταν δοκιμάσετε την επόμενη αποστολή συναλλαγής θα λάβετε InvalidPidMappingException. Ο συντονιστής συναλλαγής θα εκδώσει στη συνέχεια έναν νέο αριθμό σειράς για την επόμενη συναλλαγή. Ωστόσο, το μήνυμα ενδέχεται να χαθεί εάν το InvalidPidMappingException δεν γίνει σωστά.

Αντί για σύνολα

Όπως καταλαβαίνετε, δεν αρκεί απλώς να στέλνετε μηνύματα στον Κάφκα. Πρέπει να επιλέξετε έναν συνδυασμό παραμέτρων και να είστε έτοιμοι να κάνετε γρήγορες αλλαγές. Σε αυτό το άρθρο, προσπάθησα να δείξω λεπτομερώς τη ρύθμιση της παράδοσης ακριβώς μία φορά και περιέγραψα αρκετά προβλήματα με τις διαμορφώσεις client.id και transaksional.id που αντιμετωπίσαμε. Ακολουθεί μια σύνοψη των ρυθμίσεων Παραγωγού και Καταναλωτή.

Παραγωγός:

1. ακς = όλος
2. επαναλήψεις > 0
3. ενεργοποιώ.idempotence = αληθινός
4. max.in.flight.requests.per.connection ≤ 5 (1 για τακτική αποστολή)
5. transaksional.id = ${application-name}-${hostname}

Καταναλωτής:

1. isolation.level = read_committed

Για να ελαχιστοποιήσουμε τα σφάλματα σε μελλοντικές εφαρμογές, φτιάξαμε το δικό μας περιτύλιγμα πάνω από τη διαμόρφωση ελατηρίου, όπου οι τιμές για ορισμένες από τις παραμέτρους που παρατίθενται έχουν ήδη οριστεί.

Εδώ είναι μερικά υλικά για αυτοδιδασκαλία:

• KIP-98 - Ακριβώς μία φορά παράδοση και ανταλλαγή μηνυμάτων
• Περιγραφή ρυθμίσεων

Πηγή: www.habr.com