Problem: sifariş yazıldı, hadisə yazılmadı

Sifariş servisi sifarişi PostgreSQL-də saxlayır, sonra OrderCreated hadisəsini Kafka-ya göndərir. İki şey səhv gedə bilər. Database commit uğurlu olur, proses göndərişdən əvvəl ölür: sifariş var, amma heç kim bilmir. Ya da göndəriş uğurlu olur, commit sonra uğursuz olur: aşağı servislər mövcud olmayan sifarişə reaksiya verir. Bu dual-write problemidir: iki sistem var, ümumi transaction yoxdur, deməli atomik nəticə də yoxdur.

Həll etməzdən əvvəl mesajlaşma sisteminin nə vəd edə biləcəyini dəqiq adlandıraq.

Üç çatdırılma semantikası

  • At-most-once: mesaj sıfır və ya bir dəfə çatdırılır. Əvvəl offset commit edilir, sonra emal olunur. İkisi arasındakı çökmə mesajı itirir. İtkiyə dözümlü metrikalar üçün uyğundur, biznes hadisələri üçün nadir hallarda.
  • At-least-once: bir və ya daha çox dəfə çatdırılır. Əvvəl emal, sonra offset commit. İkisi arasındakı çökmə təkrar çatdırılmaya səbəb olur, yəni dublikat mümkündür. Praktikada default budur.
  • Exactly-once: təsir bir dəfə baş verir. Xətaları olan paylanmış sistemdə bunu təkcə şəbəkədən almaq olmur; at-least-once çatdırılma üstündə deduplikasiya və ya atomik commit ilə qurulur.

Kafka-nın exactly-once-ı nəyi əhatə edir, nəyi etmir

Kafka bir-birini tamamlayan üç mexanizm verir:

  • idempotent producer (enable.idempotence=true): broker partition daxilində producer retry-larından yaranan dublikat göndərişləri atır;
  • transaction-lar (transactional.id): producer bir neçə partition-a yazır və oxuduğu offset-ləri çıxış record-ları ilə atomik commit edir;
  • read_committed consumer-lər: ləğv edilmiş transaction-ların record-larını keçir.

Birlikdə bunlar Kafka daxilində qalan read-process-write boru xətti üçün exactly-once verir: A topic-indən oxu, çevir, B topic-inə yaz, offset-lər və çıxışlar bir vahid kimi commit olunur. Kafka Streams bunu istifadə edir.

Bu xarici side effect-lərə şamil olunmur. Consumer e-poçt göndərirsə, ödəniş API-sini çağırırsa və ya SQL bazasına yazırsa, Kafka bunu geri qaytara və ya deduplikasiya edə bilməz. Bunlar üçün idempotent consumer, nəşr tərəfində isə outbox pattern lazımdır.

Idempotent consumer

Çatdırılma at-least-once olduğuna görə handler eyni hadisəni iki dəfə görəndə eyni nəticə verməlidir. Ən sadə və etibarlı üsul işlənmiş hadisələr cədvəlidir; o, biznes dəyişikliyi ilə eyni database transaction-ında yazılır:

CREATE TABLE processed_event (    consumer   text        NOT NULL,    event_id   uuid        NOT NULL,    handled_at timestamptz NOT NULL DEFAULT now(),    PRIMARY KEY (consumer, event_id));
@KafkaListener(topics = "orders.paid", groupId = "billing")@Transactionalvoid on(ConsumerRecord<String, OrderPaid> record) {    UUID eventId = UUID.fromString(        new String(record.headers().lastHeader("event-id").value()));    int inserted = jdbc.update(""        INSERT INTO processed_event(consumer, event_id)        VALUES ('billing', ?) ON CONFLICT DO NOTHING"", eventId);    if (inserted == 0) return;          // artıq emal olunub    invoiceService.createFor(record.value());}

Insert və faktura birlikdə commit və ya rollback olur; ona görə təkrar çatdırılma ya sətri tapıb keçir, ya heç nə tapmayıb işi görür. Nəzərə alın ki, Kafka offset commit-i və database commit-i iki ayrı əməliyyatdır: listener container offset-i metod qayıtdıqdan sonra commit edir, database transaction-ı isə metod başa çatanda commit olunur və heç bir vahid transaction hər ikisini əhatə etmir. Bu ikisi arasındakı çökmə database işi artıq commit olunmuş mesajı yenidən çatdırır; təkrar çatdırılmanı təhlükəsiz edən məhz processed-events cədvəlidir. Bəzən təbii idempotentlik kifayət edir: "statusu PAID et" təkrar edilə bilər, "balansa 10 əlavə et" isə yox. HTTP üçün request açarları idempotency patterns məqaləsində izah olunub.

Transactional outbox

Dual write-ı düzəltmək üçün hadisəni vəziyyət dəyişikliyi ilə eyni lokal transaction-ın hissəsi edin. Onu outbox cədvəlinə yazın; ayrıca relay sətirləri Kafka-ya göndərir.

 +-------------- bir DB transaction --------------+
 | INSERT orders ...                              |
 | INSERT outbox (id, topic, key, payload)        |
 +------------------------------------------------+
                    |
          relay göndərilməmiş sətirləri oxuyur
                    v
          Kafka topic  --->  consumer-lər (idempotent)
CREATE TABLE outbox (    id           uuid PRIMARY KEY,    aggregate_id text        NOT NULL,    topic        text        NOT NULL,    payload      jsonb       NOT NULL,    created_at   timestamptz NOT NULL DEFAULT now(),    published_at timestamptz);CREATE INDEX outbox_unpublished ON outbox (created_at)    WHERE published_at IS NULL;
@Scheduled(fixedDelay = 500)void relay() {    for (OutboxRow r : repo.lockNextBatch(100)) { // FOR UPDATE SKIP LOCKED        kafka.send(r.topic(), r.aggregateId(), r.payload()).join();        repo.markPublished(r.id());    }}

Relay özü at-least-once-dır: göndərişdən sonra, markPublished-dən əvvəl çöksə, sətir yenidən göndərilir. Buna görə consumer-lər idempotent olmalıdır və id hadisə identifikatoru kimi header-də getməlidir. FOR UPDATE SKIP LOCKED bir neçə relay instansına cədvəli bölüşməyə imkan verir. aggregate_id ilə açar vermək Kafka-da entity üzrə sıranı saxlayır, amma bunun üçün relay eyni aggregate-in sətirlərini yaranma sırası ilə göndərməlidir.

Polling-ə alternativ database log-unu oxuyan change data capture-dır; gecikməni və yükü azaldır, amma idarə ediləcək əlavə infrastruktur komponenti gətirir.

Dead-letter topic və zəhərli mesajlar

Bəzi record-lar heç vaxt uğurlu olmayacaq: pozulmuş JSON, pozulmuş biznes qaydası, çatışmayan istinad. Sonsuz retry partition-u bloklayır. Spring Kafka-nın DefaultErrorHandler-i DeadLetterPublishingRecoverer ilə məhdud sayda backoff-lu retry edir, sonra record-u orders.paid.DLT kimi topic-ə göndərib davam edir.

@BeanDefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> t) {    var recoverer = new DeadLetterPublishingRecoverer(t);    return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3));}

Müvəqqəti xətaları (database timeout: retry) daimi xətalardan (validasiya xətası: dərhal dead-letter) ayırın. Dead-letter topic-ə monitorinq qurun, çünki izlənməyən topic məlumatı itirməyin sadəcə yavaş yoludur; record-ları kimin və necə təkrar emal edəcəyini də müəyyən edin.

Tez-tez edilən səhvlər

  • @Transactional metod daxilində kafka.send çağırıb database transaction-ının onu əhatə etdiyini güman etmək.
  • enable.idempotence-in bütün sistemi exactly-once etdiyinə inanmaq.
  • Offset-i emaldan əvvəl commit edib sonra itmiş hadisələri kəşf etmək.
  • İdempotentlik yoxlamasını side effect ilə eyni transaction-da aparmamaq.
  • Heç vaxt təmizlənməyən outbox sətirləri: cədvəl məhdudiyyətsiz böyüyür.
  • Alert və replay prosedurası olmayan dead-letter topic-lər.
  • Sabit hadisə identifikatoru olmadan relay-dən göndərmək: deduplikasiya mümkünsüz olur.

Kompromislər

Outbox cədvəl, relay və əlavə gecikmə gətirir, işlənmiş hadisələr cədvəli isə hər hadisəyə bir yazı əlavə edir. Əvəzində database ilə broker arasındakı itki pəncərəsini aradan qaldırırsınız. Bəzi bildirişin itməsi məqbuldursa, commit-dən sonra birbaşa göndərmək daha sadədir. Hadisə pul və ya stoku idarə edirsə, outbox xərci adətən özünü doğruldur.

Sıra, retry və relay

Retry-lar record-ların sırasını dəyişə bilər. Producer-in bir neçə sorğusu uçuşdadırsa və biri uğursuz olub təkrarlanırsa, sonrakı record əvvəl yerləşə bilər. Idempotent producer retry zamanı partition daxilində sıranı qoruyur; bu, onu aktiv saxlamaq üçün daha bir səbəbdir. Outbox relay-də eyni aggregate-in sətirlərini ardıcıl və yaranma sırası ilə göndərin, sətir kilidi olmadan eyni aggregate üçün iki relay işlətməyin. Müəyyən consumer üçün sıra vacib deyilsə, bunu açıq yazın: sonrakı paralellik dəyişiklikləri daha təhlükəsiz olur.

Praktik nümunə: ödəniş və faktura

Ödəniş servisi PaymentCaptured hadisəsini outbox vasitəsilə yayımlayır. Billing consumer-i hadisəni iki dəfə alırsa, processed_event cədvəlindəki primary key ikinci cəhdi dayandırır və faktura təkrar yaranmır. Bildiriş consumer-i isə xarici e-poçt provayderini çağırır; provayder idempotency açarı qəbul edirsə, hadisə id-sini həmin açar kimi ötürün, qəbul etmirsə, dublikat e-poçtun biznes üçün nə qədər zərərli olduğunu əvvəlcədən qiymətləndirin.

Yoxlama siyahısı

  1. Hər axın üçün at-least-once və ya at-most-once-ın məqbul olub-olmadığını yazın.
  2. Vacib hadisələr üçün database yazısı və göndərişi iki müstəqil addım kimi etməyin.
  3. Hadisəni vəziyyət dəyişikliyi ilə eyni transaction-da outbox-a yazın.
  4. Hər hadisəyə sabit id verin və onu header və ya sahədə daşıyın.
  5. Hər consumer-i idempotent edin, side effect ilə eyni transaction-da.
  6. Məhdud retry və monitorinqli dead-letter topic istifadə edin.
  7. Göndərilmiş outbox sətirlərini cədvəl üzrə təmizləyin.

Növbəti addım