Kafka Event Bus

The @noddde/kafka package wraps kafkajs to deliver events through Kafka topics with at-least-once semantics and per-aggregate partition ordering.

Installation

yarn add @noddde/kafka kafkajs

Configuration

import { KafkaEventBus } from "@noddde/kafka";

const eventBus = new KafkaEventBus({
  brokers: ["localhost:9092"],
  clientId: "my-service",
  groupId: "my-service-group",
  topicPrefix: "noddde.", // optional — topics become "noddde.AccountCreated"
  sessionTimeout: 60000, // optional — increase for slow handlers
  heartbeatInterval: 5000, // optional — must be < sessionTimeout / 3
  resilience: { maxAttempts: 11, initialDelayMs: 500, maxRetries: 5 }, // optional — connection + delivery resilience
  partitionKeyStrategy: "aggregateId", // optional — default uses aggregateId for per-aggregate ordering
  // logger: myCustomLogger, // optional — defaults to NodddeLogger("warn", "noddde:kafka")
});

How It Works

• Publishing: Events are serialized as JSON and sent to a Kafka topic derived from the event name (${topicPrefix}${event.name}). The message key is determined by the partitionKeyStrategy option: by default ("aggregateId"), the key is event.metadata?.aggregateId (stringified), ensuring per-aggregate partition ordering. If no aggregateId is present, the key is null (round-robin). A custom function can be provided to derive the key from the full event.
• Subscribing: Handlers registered via on() receive messages through a Kafka consumer. Handlers registered before connect() are buffered and subscriptions are created when the connection is established. Multiple handlers for the same event are invoked in parallel via Promise.allSettled() — every handler runs to completion even when some fail, and each failure is logged individually with structured fields (see Handler Error Isolation).
• Delivery: At-least-once. The consumer runs with autoCommit: false — offsets are committed explicitly via consumer.commitOffsets() only after all handlers complete successfully. If any handler fails, the first rejection is re-thrown after all siblings have settled (so the offset is not committed and the message is redelivered). After a successful commit, the in-memory delivery count entry for the message is pruned to prevent unbounded memory growth. Consumers must be idempotent — handlers that completed before a sibling failure will re-execute on redelivery.
• Poison message protection: Malformed messages that fail JSON deserialization are logged and skipped (offset committed), preventing corrupt data from blocking the partition. If resilience.maxRetries is configured, messages that exceed the delivery attempt limit are also skipped with a warning.
• Connection resilience: The optional resilience config (BrokerResilience from @noddde/core) is mapped to kafkajs retry options for automatic reconnection on broker unavailability. maxAttempts maps to kafkajs retries (minus 1, since kafkajs counts retries after the first attempt), initialDelayMs to initialRetryTime, and maxDelayMs to maxRetryTime. The sessionTimeout and heartbeatInterval options tune consumer rebalance behavior for slow handlers. Concurrent connect() calls are deduplicated via a connection promise mutex to prevent parallel connection attempts.
• Logging: All internal logging uses the framework Logger interface from @noddde/core. Pass a custom logger in the config or accept the default (NodddeLogger("warn", "noddde:kafka")). All log calls include structured context data (event name, error details).

Cold-start warmup

A freshly-deployed Kafka cluster is slow on its first end-to-end publish/consume cycle: topic auto-creation, leader election, and ISR sync happen lazily on first use, so the first message can take far longer to be delivered than every message after it. connect() waits for the consumer to begin polling (via kafkajs FETCH_START), but that wait does not cover this broker-side cold start.

warmup() addresses it by running a throwaway publish/consume round-trip on a dedicated internal topic (named from clientId) and resolving only once the round-trip is observed:

await eventBus.connect();
await eventBus.warmup(); // resolves once the broker has served a full round-trip

• Must follow connect(): calling warmup() before connect() or after close() throws the same "not connected" error as dispatch().
• Idempotent: after the first successful call, subsequent calls resolve immediately without repeating the round-trip. Concurrent overlapping calls are deduplicated into a single round-trip via an in-flight promise mutex (mirroring connect()).
• Bounded: rejects with a timeout error if the round-trip does not complete within warmupTimeoutMs (default 60000) rather than hanging indefinitely.

Set warmupOnConnect: true to fold the warmup into connection so a single await connect() returns a fully warmed bus. A warmup failure then propagates through connect()'s returned promise, consistent with how connect() surfaces other connection errors:

const eventBus = new KafkaEventBus({
  brokers: ["localhost:9092"],
  clientId: "my-service",
  groupId: "my-service-group",
  warmupOnConnect: true,
  warmupTimeoutMs: 60000, // optional — round-trip timeout (default 60000)
});

await eventBus.connect(); // only resolves once the broker is warm

Wiring with Domain

import { wireDomain } from "@noddde/engine";
import { KafkaEventBus } from "@noddde/kafka";

const domain = await wireDomain(myDomain, {
  buses: () => ({
    eventBus: new KafkaEventBus({
      brokers: ["localhost:9092"],
      clientId: "my-service",
      groupId: "my-service-group",
    }),
  }),
});

The Domain auto-calls connect() on the event bus during wiring (via Connectable auto-discovery) and close() on shutdown (via Closeable auto-discovery). No manual lifecycle management needed.