Kafka fejlesztői és üzemeltetési útmutató
A MailGen Kafka-rétege tartós háttérmunkát és mailbox-sorrendhez kötött feldolgozást támogat. Ez a referencia a jelenlegi újrahasználható magot és a KafkaTester-gyakorlati runtime-ot fedi le. A teljes Docker-alkalmazáshoz használd az első setupot.
1. Kafka-fogalmak
A producer rekordokat ír egy topicba. A brokerek rendezett partíciókban tárolják őket. A consumerek consumer groupokon keresztül olvassák azokat. Minden partíciónak saját offset-pozíciói vannak. A rendezettség egy partíción belül érvényes, nem globálisan topicok vagy partíciók között.
A kulcs a routing és az üzleti határ azonosítója. A MailGen mailbox-publikálás emellett megadja a hozzárendelt explicit partíciót. A kézbesítési modellekért és a group-viselkedésért olvasd el az alapokat.
2. Megvalósítási határok
Az újrahasználható mag a src/lib/kafka mappában van. A KafkaTester domain-orchestrációja a src/services/KafkaTester mappában van. A megosztott típusok a src/lib/types/kafka mappában vannak.
A mag elemei közé tartozik a KafkaCoreServiceBase, a KafkaProducerClient, a KafkaJsonConsumerService, a topic- és partícióadminisztráció, az outbox queue, publisher és tracing, a deduplikáció és a szekvenciális hozzárendelés.
A Prisma a MailboxUser, a SequentialTopicPartition, az OutboxEvent és a ProcessedKafkaMessage modelleket a kafka sémára képezi le. Őrizd meg ezeket a leképezéseket és állapotaikat a megfelelő broker-adatokkal együtt.
4. Könyvtárreferencia
| Backend-útvonal | Cél |
|---|---|
src/lib/kafka/env.ts | Környezet betöltése és validálása |
src/lib/kafka/config.ts | Admin- és alkalmazás-kliens konfigurációk előállítása |
src/lib/kafka/producer.ts | JSON küldése opcionális explicit partícióval |
src/lib/kafka/consumer.ts | JSON dekódolása, handler hívása, commit utána |
src/lib/kafka/partition-admin.ts | Broker-topicok ellenőrzése, létrehozása és bővítése |
src/lib/kafka/topics.ts | Topic-regiszter |
src/lib/kafka/core-service.ts | Közös életciklus |
src/lib/kafka/patterns/ | Újrahasználható outbox, idempotencia és partícióhozzárendelés |
src/services/KafkaTester/runKT.ts | HTTP/service belépési pont |
src/services/KafkaTester/events/mailbox-event.ts | Eseményszerződés és factory-k |
src/services/KafkaTester/modules/runtime/kafka-tester.service.ts | Gyakorlati runtime-orchestráció |
src/services/KafkaTester/modules/mailbox/ | Felhasználó-, projekció-, feldolgozó- és workerlogika |
5. Önálló fejlesztői setup
5.1 Előfeltételek
Használj Node.js 22 vagy újabb verziót, a rögzített pnpm 10.27.0 verziót, elérhető, hitelesített Kafkát, valamint PostgreSQL-t a szükséges Prisma-sémákkal. A natív install és a futtatás során ugyanazt a Node-főverziót használd.
Válassz elkülönített gyakorlati adatbázist és streamet a KafkaTesterhez. Parancsai felhasználókat, eseményeket és partíciókat hoznak létre. Nem csak olvasható éles diagnosztikai eszköz.
5.2 Környezet
| Beállítás | Jelentés és formátum | Szükséges kontextus vagy példa |
|---|---|---|
KAFKA_CLIENT_ID | Stabil kliens-címke | Példa: ecs-backend |
KAFKA_NETWORK | lan vagy tailscale endpoint-kiválasztás | A kiválasztott hálózatnak elérhetőnek kell lennie |
KAFKA_LAN_BOOTSTRAP_SERVERS, KAFKA_TAILSCALE_BOOTSTRAP_SERVERS | Vesszővel elválasztott host:port listák | Mindkét listának validnak kell lennie. Példa helyi IPv4-re: 127.0.0.1:9092,... |
KAFKA_SECURITY_PROTOCOL | Broker-transport | Hitelesített, privát plaintext csak ott, ahol a szabályzat engedi. Nyilvános transportnál használj TLS-t. |
KAFKA_SASL_MECHANISM | Hitelesítési mechanizmus | Egyezzen a broker-konfigurációval, például PLAIN |
KAFKA_ADMIN_USERNAME, KAFKA_ADMIN_PASSWORD | Adminisztrációs szerep | Topic- és partícióműveletekhez kötelező |
KAFKA_APP_USERNAME, KAFKA_APP_PASSWORD | Alkalmazási szerep | Normál producer és consumer |
KAFKA_TOPIC_USER_CREATED, KAFKA_TOPIC_USER_CREATED_DLQ | Felhasználó-létrehozási esemény és dead-letter topic | Példa: app.user.created.v1 és a hozzá tartozó .dlq |
MAX_USERS_PER_QueProcessor_PARTITION | Pozitív limit felhasználóra vagy mailboxra partíciónként | Alapértelmezett 1, egész szám tartomány: 1–100. A backend hozzárendelés ezt használja. Csak a hozzárendelési szabályzattal együtt módosítsd. |
Az egységes Docker-runtime a gyökér környezetből származtatja a szolgáltatásbeállításokat. Ne cseréld le a gyökér bemeneteket önálló példákra. A valódi távoli mód szigorúbb transport- és identitás-ellenőrzéseket igényel.
5.3 Install
cd mailGen/msmicroservices-architexture
pnpm install --trust-lockfile
Csak a megbízható, commitolt lockfile esetén használd a --trust-lockfile kapcsolót. A postinstall a Kafka natív addonját az aktív Node-verzióhoz építi újra, és megkísérli a better-sqlite3 újraépítését. Az olyan ABI-eltérés, mint a 115-ös és a 127-es modulverzió, egyező Node- és install-kombinációt igényel. A pnpm kafka:rebuild-native parancs explicit módon újrafuttatja a natív újraépítést.
6. KafkaTester runtime
Watch módhoz futtasd a pnpm dev:kt parancsot. Várj Fastify listen üzenetet, Kafka-kapcsolatot, partíció-hozzárendelést és időszakos outbox idle naplókat.
Önálló indítás előtt állítsd be a KAFKA_TESTER_PORT vagy a PORT változót. A dokumentált példa a 3011 portot használja, de a belépési pontnak nincs implicit portalapértelmezése. A 3011-es porttal:
curl http://127.0.0.1:3011/healthcheck
Az egészséges válasz:
{"success":true,"service":"KafkaTester","status":"ok"}
A konfigurált portfelülírás megváltoztatja ezt a címet. Az egészség-ellenőrzés önmagában nem bizonyítja az eseményfeldolgozást.
Lefordított futtatáshoz:
pnpm build
pnpm start:kt
Ugyanez a runtime a dist belépési pontját használja. Felosztott consumer, dispatcher és interaktív producer esetén kövesd a CLI-parancsokat. Ne kombináld véletlenül a dev:kt paranccsal ugyanazon a gyakorlaton.
A createNewSequentialPartition() és a listSequentialPartitions() explicit adminisztrációt tesz elérhetővé egy gyakorlat partíció-elosztásához. Mindkettő tartós állapotot és broker-elrendezést módosít.
Sorrend és deduplikáció ellenőrzése
Indíts egy szándékolt topológiát, hozz létre egy gyakorlati felhasználót, majd állítsd sorba és publikáld a szekvenciáját. Ellenőrizd a hozzárendelési naplókat, a mailbox-feldolgozást és az állapotátmeneteket. Vizsgáld a MailboxUser.lastSequence értéket, az outbox feldolgozási és publikálási állapotát, valamint a ProcessedKafkaMessage sorokat. Ne következtess sikerre pusztán a sorba állított események számából.
SecondBun parancsmegfeleltetés
| Gyakorlati bemenet | KafkaTester-művelet |
|---|---|
c | createMailboxUser() |
Felhasználói azonosító, például 1 | queueAllMailboxEventsForUser(userId), majd publishOutboxBatch() |
q<id> | queueAllMailboxEventsForUserWithStatus(...PREQUEUED) |
s<id> | releasePrequeuedOutboxEventsByKey(String(userId)) a src/lib/kafka/index.ts fájlból |
r<id> | A kiválasztott sikertelen outbox-események újrapróbálása ehhez a kulcshoz: retryFailedOutboxEventsByKey(String(userId)) |
p | createNewSequentialPartition() |
l | listMailboxUsers() és listSequentialPartitions() |
7. Eseménymodell és routing
A mailbox-gyakorlat eseménytípusai: mail.new, mail.delete, folder.create és folder.delete. Közös borítékjuk a következőket tartalmazza: eventId, eventVersion, occurredAt, userId és sequence.
A publikálás a key = String(userId) értéket és a felhasználó dedikált explicit partícióját adja meg. A tartós hozzárendelés, nem csak az alapértelmezett hash-particionáló, hozza létre a felhasználóhoz kötött routingot. A consumer a projekcióban ellenőrzi az üzleti sorrendet.
Az aktuális mailbox-parancsok routingja a kanonikus mailbox-felhasználói azonosítót használja a msintegrationv2.users.user_id mezőből. A böngészőben bejelentkezett szereplő vagy a megjelenített e-mail másik azonosító.
8. Outbox-állapotreferencia
| Állapot | Jelentés |
|---|---|
PREQUEUED | Explicit kiadásig tartva |
PENDING | Kész a dispatch-lefoglalásra |
DISPATCHING | Lefoglalt publikálás folyamatban |
PUBLISHED | A publikálás rögzítve |
UNDER_PROCESSING | Consumer-feldolgozás folyamatban |
SUCCESS | A feldolgozás véget ért |
PUSHED_FAILURE | A consumer hibás útja rögzítve |
FAILURE | Publikálási vagy hibaállapot |
A publisher atomikusan lefoglalja a függő munkát. Feltételesen jelöli a publikálást, és nem írja felül az előrehaladott feldolgozási eredményeket. A consumer elindulhat a DISPATCHING állapotból, ha megnyeri a versenyt a publikálási visszaigazolással. A végállapot-ellenőrzések megakadályozzák a szokásos visszafelé mozgást.
A stale dispatch-helyreállítás a broker állapotát vizsgálja, mielőtt eldöntené a publikálás lezárását vagy a visszasorolást. Az ismeretlen partíció hibáinak helyreállítása külön szabályozott. Az általános átmeneti segéd a SUCCESS, a PUSHED_FAILURE és a FAILURE állapotot végállapotnak tekinti. A kifejezett helyreállítás viszont kiválasztott sikertelen sorokat is kezelhet.
A claimPendingEventForDispatch, a markPublished, a markPublishFailed és a requeueStaleDispatchingEvents függvények a patterns/outbox/outbox-publisher.ts fájlban vannak. A közös nyomkövetés az outbox-tracing.ts fájlban van. A KafkaTester mailbox-feldolgozása rögzíti a consumer állapotát. Lásd a mintákat.
9. Idempotencia
A processKafkaMessageOnce a group/topic/partition/offset azonosítót használja egyedi azonosítóként. Beszúrja a deduplikációs rekordot, és egy tranzakcióban futtatja a work(tx) hívást. Egy duplikátum az already_processed értéket adja vissza. Más hibák tovább terjednek.
Ez a védelem az adott azonosító tranzakciós adatbázis-munkáját védi. Nem teszi pontosan egyszer végrehajthatóvá a tetszőleges külső mellékhatásokat. Nem deduplikál különböző offseteket vagy groupokat üzleti eseményazonosító alapján.
10. Consumer commit-viselkedés
A consumer kikapcsolja az automatikus commitot. Sikeres handler után explicit módon commitálja az offset + 1 értéket. A commit előtti hiba újrafeldolgozáshoz vezethet. Az üres értékeket kihagyja, illetve commitálja, hacsak a hívó nem engedélyezi kifejezetten őket.
A jelenlegi kliens-alapértelmezések közé tartozik a cooperative-sticky hozzárendelés. Ide tartozik még a 45 másodperces session timeout, a 3 másodperces heartbeat és a 300,000 ms maximális poll-intervallum. Az átadott poll-intervallumnak 1 és 43,200,000 ms közötti biztonságos egész számnak kell lennie.
A producer idempotenciát, az acks=all beállítást, legfeljebb öt in-flight kérést, 120,000 ms delivery timeoutot és 30,000 ms request timeoutot használ. Ezek a beállítások nem helyettesítik az adatbázis- vagy provider-idempotenciát.
11. Dead-letter kezelés
A dead-letter queue (DLQ) a feldolgozási hibákat tárolja vizsgálatra. A handler megadja az eredeti topic, partíció és offset metaadatot, az okot és a payloadot. Védd az érzékeny payloadokat. A DLQ-rekord hibaút, nem sikeres üzleti lezárás.
14. Üzemeltetési ellenőrzések
Készítsd elő a topicokat az ellenőrzött admin- és indítási úton. Tartsd explicit a replikációs, megtartási és hozzáférési szabályzatokat. Figyeld a consumer lagot, a DLQ-forgalmat, az outbox-torlódást és a rebalance-eseményeket. Ellenőrizd a partícióbővítést a tartós hozzárendelésekkel szemben. Tartsd konzisztensen a Node-verziókat a natív buildeknél.
Az elfogadott vagy publikált mailbox-munkának továbbra is szüksége van a művelet provider-megerősítésére. Használd az alkalmazás műveleti állapotát és a kanonikus visszaolvasást.
15. Hibaelhárítás
Hibaelhárításhoz használd a hibaelhárítási útmutatót: ABI-hibák, Corepack-aláírások, hiányzó hozzárendelések, elakadt outbox-munka, érvénytelen átmenetek és group-eloszlás esetén. A diagnózis során őrizd meg az offseteket és a deduplikációs állapotot. A topicok vagy adatbázissorok alaphelyzetbe állítása destruktív karbantartás, nem kapcsolati javítás.