KafkaTester CLI usage
Use KafkaTester only with a prepared, isolated exercise environment. These commands create users, partition assignments, and events in its configured database/broker. Do not point them at production mailbox state.
Run from mailGen/msmicroservices-architexture after the developer prerequisites. Choose one topology: all-in-one or split workers.
Mode 1: all-in-one
pnpm dev:kt
This starts the HTTP health endpoint, mailbox consumer, and outbox dispatcher. Expect listen, connection, assignment, and periodic idle logs. A health response alone does not prove message processing.
Mode 2: split workers
Open three terminals in the backend directory.
| Terminal | Command | Role |
|---|---|---|
| A | pnpm kafka:kt:consume:mailbox-events | Mailbox consumer |
| B | pnpm kafka:kt:outbox:watch | Outbox dispatcher |
| C | pnpm kafka:kt:producer:daemon | Interactive producer |
Do not also run pnpm dev:kt against that same exercise state. Overlap complicates dispatch/consumption and status ownership.
Producer commands
| Input | Action |
|---|---|
c | Create a mailbox user |
p | Create a sequential partition |
l | List users and partitions |
q<id> | Queue PREQUEUED events for that user without publishing |
s<id> | Release that user's PREQUEUED events to PENDING |
r<id> | Return selected failed outbox events for that user to PENDING for retry |
b | Publish one outbox batch |
<id> | Queue and publish the mailbox-event sequence for that user |
q | Quit |
Use an existing exercise user ID from l. Invalid input does not start a mailbox simulation. Queue creation and batch publication have separate result counts.
Multiple consumers
For partition sharing, run pnpm kafka:kt:consume:mailbox-events in two terminals with the same group. Expect assignments to divide available partitions. Idle consumers can be correct when partitions are fewer than members.
For independent groups:
KAFKA_TESTER_CONSUMER_GROUP=bun-mailbox-events-worker pnpm kafka:kt:consume:mailbox-events
In another terminal:
KAFKA_TESTER_CONSUMER_GROUP=bun-mailbox-events-worker-2 pnpm kafka:kt:consume:mailbox-events
Each group reads the stream independently. This does not mean the shared domain projection can apply a business sequence twice. Group name changes also change deduplication identity.
One-shot producer examples
Create an exercise user:
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); });"
Queue and publish for user 1. Replace it with an actual exercise user ID:
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); });"
Create, prequeue, release, and publish:
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); });"
Check the result
Inspect the user/partition list, event processing logs, MailboxUser.lastSequence, outbox status, and ProcessedKafkaMessage rows. Publication count alone does not prove processing. Use troubleshooting for stalled work. Do not delete topics or deduplication state to repair it.