KafkaTester CLI-használat
A KafkaTestert csak előkészített, elkülönített gyakorlati környezetben használd. Ezek a parancsok felhasználókat, partíció-hozzárendeléseket és eseményeket hoznak létre a konfigurált adatbázisban és brokerben. Ne irányítsd őket éles mailbox-állapotra.
Futtasd a parancsokat a mailGen/msmicroservices-architexture mappából, miután teljesítetted a fejlesztői előfeltételeket. Válassz egy topológiát: all-in-one vagy felosztott workerek.
1. mód: all-in-one
pnpm dev:kt
Ez a HTTP health endpointot, a mailbox consumert és az outbox dispatchert indítja el. Várj listen, kapcsolódási, hozzárendelési és időszakos idle naplókat. Egyedül a health-válasz nem bizonyítja az üzenetfeldolgozást.
2. mód: felosztott workerek
Nyiss három terminált a backend mappájában.
| Terminál | Parancs | Szerep |
|---|---|---|
| A | pnpm kafka:kt:consume:mailbox-events | Mailbox consumer |
| B | pnpm kafka:kt:outbox:watch | Outbox dispatcher |
| C | pnpm kafka:kt:producer:daemon | Interaktív producer |
Ne futtasd emellett a pnpm dev:kt parancsot ugyanazon a gyakorlati állapoton. Az átfedés bonyolítja a dispatch, a fogyasztás és a státuszfelelősség kezelését.
Producer-parancsok
| Bemenet | Művelet |
|---|---|
c | Mailbox-felhasználó létrehozása |
p | Szekvenciális partíció létrehozása |
l | Felhasználók és partíciók listázása |
q<id> | Az adott felhasználó PREQUEUED eseményeinek sorba állítása publikálás nélkül |
s<id> | Az adott felhasználó PREQUEUED eseményeinek kiadása PENDING állapotba |
r<id> | Az adott felhasználó kiválasztott, sikertelen outbox-eseményeinek visszaállítása PENDING állapotba újrapróbálkozáshoz |
b | Egy outbox-köteg publikálása |
<id> | Az adott felhasználó mailbox-eseménysorozatának sorba állítása és publikálása |
q | Kilépés |
Használj létező gyakorlati felhasználói azonosítót az l listából. Az érvénytelen bevitel nem indít mailbox-szimulációt. A sorba állítás és a köteg-publikálás külön darabszámot ad vissza.
Több consumer
Partíciómegosztás teszteléséhez futtasd a pnpm kafka:kt:consume:mailbox-events parancsot két terminálban ugyanazzal a grouppal. A hozzárendelések elosztják az elérhető partíciókat. A tétlen consumer helyes lehet, ha kevesebb a partíció, mint a tag.
Független groupokhoz:
KAFKA_TESTER_CONSUMER_GROUP=bun-mailbox-events-worker pnpm kafka:kt:consume:mailbox-events
Egy másik terminálban:
KAFKA_TESTER_CONSUMER_GROUP=bun-mailbox-events-worker-2 pnpm kafka:kt:consume:mailbox-events
Mindegyik group függetlenül olvassa a folyamot. Ez nem jelenti azt, hogy a közös domain-projekció kétszer alkalmazhat ugyanazt az üzleti sorrendet. A group nevének módosítása a deduplikációs azonosítót is megváltoztatja.
Egylépéses producer-példák
Hozz létre egy gyakorlati felhasználót:
pnpm exec tsx -e "import { KafkaTesterService } from './src/services/KafkaTester/modules/runtime/kafka-tester.service.ts'; (async () => { const s = new KafkaTesterService(); const u = await s.createMailboxUser(); console.log(JSON.stringify({ userId: u.id, partition: u.kafkaPartition })); process.exit(0); })().catch((e) => { console.error(e); process.exit(1); });"
Sorba állítás és publikálás a 1 azonosítójú felhasználóhoz. Cseréld le egy tényleges gyakorlati felhasználói azonosítóra:
pnpm exec tsx -e "import { KafkaTesterService } from './src/services/KafkaTester/modules/runtime/kafka-tester.service.ts'; (async () => { const s = new KafkaTesterService(); const events = await s.queueAllMailboxEventsForUser(1); const published = await s.publishOutboxBatch(); console.log(JSON.stringify({ queued: events.length, published })); process.exit(0); })().catch((e) => { console.error(e); process.exit(1); });"
Létrehozás, sorba állítás, kiadás és publikálás:
pnpm exec tsx -e "import { KafkaTesterService } from './src/services/KafkaTester/modules/runtime/kafka-tester.service.ts'; import { OUTBOX_EVENT_STATUS, releasePrequeuedOutboxEventsByKey } from './src/lib/kafka/index.ts'; (async () => { const s = new KafkaTesterService(); const u = await s.createMailboxUser(); await s.queueAllMailboxEventsForUserWithStatus({ userId: u.id, status: OUTBOX_EVENT_STATUS.PREQUEUED }); const released = await releasePrequeuedOutboxEventsByKey(String(u.id)); const published = await s.publishOutboxBatch(); console.log(JSON.stringify({ userId: u.id, released, published })); process.exit(0); })().catch((e) => { console.error(e); process.exit(1); });"
Az eredmény ellenőrzése
Ellenőrizd a felhasználó- és partíciólistát, az eseményfeldolgozási naplókat, a MailboxUser.lastSequence értéket, az outbox státuszát és a ProcessedKafkaMessage sorokat. Egyedül a publikálási darabszám nem bizonyítja a feldolgozást. Elakadt munka esetén használd a hibaelhárítási útmutatót. Ne törölj topicokat és deduplikációs állapotot a javításhoz.