Kafka developer and operations guide
MailGen's Kafka layer supports durable background work and mailbox-ordered processing. This reference covers the current reusable core and KafkaTester exercise runtime. Use first installation for the full Docker application.
1. Kafka concepts
A producer writes records to a topic. Brokers store them in ordered partitions. Consumers read through consumer groups. Each partition has its own offset positions. Ordering holds within one partition, not globally across topics or partitions.
A key identifies the routing/business boundary. MailGen mailbox publication also supplies an explicit assigned partition. Read fundamentals for delivery models and group behavior.
2. Implementation boundaries
Reusable core lives in src/lib/kafka. KafkaTester domain orchestration lives in src/services/KafkaTester. Shared types live in src/lib/types/kafka.
Core elements include KafkaCoreServiceBase, KafkaProducerClient, KafkaJsonConsumerService, topic/partition administration, outbox queue/publisher/tracing, deduplication, and sequential assignment.
Prisma maps MailboxUser, SequentialTopicPartition, OutboxEvent, and ProcessedKafkaMessage to the kafka schema. Preserve their mapping and state with the corresponding broker data.
4. Directory reference
| Backend path | Purpose |
|---|---|
src/lib/kafka/env.ts | Load and validate environment |
src/lib/kafka/config.ts | Produce admin/application client configurations |
src/lib/kafka/producer.ts | Send JSON with an optional explicit partition |
src/lib/kafka/consumer.ts | Decode JSON, call handler, commit afterward |
src/lib/kafka/partition-admin.ts | Inspect/create/expand broker topics |
src/lib/kafka/topics.ts | Topic registry |
src/lib/kafka/core-service.ts | Shared lifecycle |
src/lib/kafka/patterns/ | Reusable outbox, idempotency, partition assignment |
src/services/KafkaTester/runKT.ts | HTTP/service entrypoint |
src/services/KafkaTester/events/mailbox-event.ts | Event contract and factories |
src/services/KafkaTester/modules/runtime/kafka-tester.service.ts | Exercise runtime orchestration |
src/services/KafkaTester/modules/mailbox/ | User, projection, processing, and worker logic |
5. Standalone developer setup
5.1 Prerequisites
Use Node.js 22 or newer, pinned pnpm 10.27.0, reachable authenticated Kafka, and PostgreSQL with the required Prisma schemas. Use the same Node major for native installation and execution.
Choose an isolated exercise database/stream for KafkaTester. Its commands create users, events, and partitions. It is not a read-only production diagnostic.
5.2 Environment
| Setting | Meaning and format | Required context/example |
|---|---|---|
KAFKA_CLIENT_ID | Stable client label | Example ecs-backend |
KAFKA_NETWORK | lan or tailscale endpoint selection | Selected network must be reachable |
KAFKA_LAN_BOOTSTRAP_SERVERS, KAFKA_TAILSCALE_BOOTSTRAP_SERVERS | Comma-separated host:port lists | Both lists must validate. Example local IPv4 127.0.0.1:9092,... |
KAFKA_SECURITY_PROTOCOL | Broker transport | Authenticated private plaintext only where policy permits. Use TLS for public transport. |
KAFKA_SASL_MECHANISM | Authentication mechanism | Match broker configuration, for example PLAIN |
KAFKA_ADMIN_USERNAME, KAFKA_ADMIN_PASSWORD | Administration role | Required for topic/partition work |
KAFKA_APP_USERNAME, KAFKA_APP_PASSWORD | Application role | Normal producer/consumer |
KAFKA_TOPIC_USER_CREATED, KAFKA_TOPIC_USER_CREATED_DLQ | User-created event and dead-letter topic | Example app.user.created.v1 and matching .dlq |
MAX_USERS_PER_QueProcessor_PARTITION | Positive users/mailboxes per partition limit | Default 1, integer range 1–100. Backend assignment consumes it. Change with assignment policy, not alone. |
Unified Docker derives service settings from the root environment. Do not replace root inputs with standalone examples. Real remote mode has stricter transport/identity checks.
5.3 Install
cd mailGen/msmicroservices-architexture
pnpm install --trust-lockfile
Use --trust-lockfile only for the trusted committed lockfile. postinstall rebuilds Kafka's native addon for the active Node version and attempts better-sqlite3 rebuilding. An ABI mismatch such as module version 115 versus 127 needs the matching Node/install combination. pnpm kafka:rebuild-native reruns the native rebuild explicitly.
6. KafkaTester runtime
For watch mode, run pnpm dev:kt. Expect Fastify listen, Kafka connection, partition assignment, and periodic outbox idle logs.
Set KAFKA_TESTER_PORT or PORT before standalone startup. The commented example uses 3011, but the entrypoint has no implicit port default. With port 3011:
curl http://127.0.0.1:3011/healthcheck
Its healthy response is:
{"success":true,"service":"KafkaTester","status":"ok"}
A configured port override changes that address. Health alone does not prove event processing.
For compiled execution:
pnpm build
pnpm start:kt
The same runtime uses its dist entrypoint. For split consumer/dispatcher/interactive producer, follow CLI commands. Do not accidentally combine it with dev:kt against one exercise.
createNewSequentialPartition() and listSequentialPartitions() expose explicit administration for an exercise's partition distribution. They change persistent state and broker layout.
Check ordering and deduplication
Start one intended topology, create an exercise user, and queue/publish its sequence. Check assignment logs, mailbox processing, and status transitions. Inspect MailboxUser.lastSequence, outbox processing/publication state, and ProcessedKafkaMessage rows. Do not infer success from queue count alone.
SecondBun command correspondence
| Exercise input | KafkaTester operation |
|---|---|
c | createMailboxUser() |
User ID, such as 1 | queueAllMailboxEventsForUser(userId) then publishOutboxBatch() |
q<id> | queueAllMailboxEventsForUserWithStatus(...PREQUEUED) |
s<id> | releasePrequeuedOutboxEventsByKey(String(userId)) from src/lib/kafka/index.ts |
r<id> | Retry selected failed outbox events for that key with retryFailedOutboxEventsByKey(String(userId)) |
p | createNewSequentialPartition() |
l | listMailboxUsers() and listSequentialPartitions() |
7. Event model and routing
Mailbox exercise event types are mail.new, mail.delete, folder.create, and folder.delete. Their shared envelope contains eventId, eventVersion, occurredAt, userId, and sequence.
Publication supplies key = String(userId) and the user's dedicated explicit partition. The persistent assignment, rather than only a default hash partitioner, establishes user-scoped routing. The consumer checks business sequence in its projection.
Actual mailbox command routing uses the canonical mailbox user identity from msintegrationv2.users.user_id. A browser actor or display email is a different identity.
8. Outbox state reference
| State | Meaning |
|---|---|
PREQUEUED | Held until explicit release |
PENDING | Ready for dispatch claim |
DISPATCHING | Claimed publication in progress |
PUBLISHED | Publication recorded |
UNDER_PROCESSING | Consumer processing |
SUCCESS | Processing completed |
PUSHED_FAILURE | Consumer failure path recorded |
FAILURE | Publication/failure state |
The publisher claims pending work atomically. It marks publication conditionally and does not overwrite advanced processing results. A consumer can start from DISPATCHING if it wins the race with publication acknowledgment. Terminal-state checks prevent ordinary backward transitions.
Stale dispatch recovery probes broker state before choosing publication completion or requeue. Unknown-partition failure recovery is separately controlled. The ordinary transition helper treats success/pushed-failure/failure as terminal, while explicit recovery can address selected failed rows.
Functions claimPendingEventForDispatch, markPublished, markPublishFailed, and requeueStaleDispatchingEvents live in patterns/outbox/outbox-publisher.ts. Shared tracing lives in outbox-tracing.ts. KafkaTester mailbox processing records consumer state. See patterns.
9. Idempotency
processKafkaMessageOnce uses group/topic/partition/offset as its unique identity. It inserts the deduplication record and executes work(tx) in one transaction. A duplicate returns already_processed. Other failures propagate.
This protects transactional database work for the same identity. It does not make arbitrary external side effects exactly-once, or deduplicate different offsets/groups by business event ID.
10. Consumer commit behavior
The consumer disables automatic commits. After a successful handler, it explicitly commits offset + 1. A failure before commit can cause reprocessing. Empty values are skipped/committed unless the caller explicitly enables them.
Current client defaults include cooperative-sticky assignment, 45-second session timeout, 3-second heartbeat, and 300,000 ms maximum poll interval. A supplied poll interval must be a safe integer from 1 to 43,200,000 ms.
The producer uses idempotence, acks=all, up to five in-flight requests, 120,000 ms delivery timeout, and 30,000 ms request timeout. These settings do not replace database/provider idempotency.
11. Dead-letter handling
A dead-letter queue (DLQ) stores processing failures for inspection. The handler supplies original topic/partition/offset metadata, reason, and payload. Protect sensitive payloads. A DLQ record is a failure path, not successful business completion.
14. Operational checks
Prepare topics through the controlled admin/startup path. Keep replication, retention, and access policies explicit. Monitor consumer lag, DLQ traffic, outbox backlog, and rebalance events. Check partition expansion against persistent assignments. Keep Node versions consistent across native builds.
Accepted or published mailbox work still needs its operation's provider confirmation. Use the application operation state and canonical readback.
15. Troubleshooting
Use the troubleshooting guide for native ABI, Corepack signatures, absent assignments, stalled outbox work, invalid transitions, and group distribution. Preserve offsets/deduplication state during diagnosis. Resetting topics or database rows is destructive maintenance, not a connection fix.