@drarzter/kafka-client

Type-safe Kafka client for Node.js. Framework-agnostic core with a first-class NestJS adapter. Built on top of @confluentinc/kafka-javascript (librdkafka).

🕹️ Try it live — kafka-playground: an interactive sandbox that spins up Kafka + this library in Docker and lets you create producers/consumers from a dashboard and watch retries, DLQ, circuit breaker, delayed delivery, dedup (Redis), transactional outbox (Postgres), and Avro/Protobuf serde work in real time. git clone && docker compose up.

Table of contents

• What is this?
• Why?
• How it compares
• Installation
• Standalone usage
• Quick start (NestJS)
• Usage
  • 1. Define your topic map
  • 2. Register the module
  • 3. Inject and use
• Consuming messages
  • Declarative: @SubscribeTo()
  • Imperative: startConsumer()
  • Regex topic subscription
  • Iterator: consume()
• Multiple consumer groups
• Partition key
  • Typed partition keys
• Message headers
• Batch sending
• Delayed delivery
• Batch consuming
• Tombstone messages
• Compression
• Transactions
• Consumer interceptors
• Instrumentation
  • OpenTelemetry metrics
• Transport security
  • AWS MSK IAM & GCP authentication
  • ACL requirements
• Environment configuration
• Options reference
• Error classes
• Deduplication (Lamport Clock)
  • Pluggable deduplication store
• Retry topic chain
• stopConsumer
• Pause and resume
• Circuit breaker
  • getCircuitState
• Reset consumer offsets
• Seek to offset
• Seek to timestamp
• Message TTL
• DLQ replay
• Read snapshot
• Offset checkpointing
  • checkpointOffsets
  • restoreFromCheckpoint
• Windowed batch consumer
• Header-based routing
• Lag-based producer throttling
• Transactional consumer
• Transactional outbox
• Serialization: JSON, Avro, Protobuf
• Schema Registry client
• Admin API
• DLQ CLI
• Graceful shutdown
• Consumer handles
• onMessageLost
• onTtlExpired
• onRebalance
• Consumer lag
• Handler timeout warning
• Static group membership
• Schema validation
  • Versioned schemas
  • Context-aware validators
• Constructor options validation
• Health check
• Testing
• Project structure

What is this?

An opinionated, type-safe abstraction over @confluentinc/kafka-javascript (librdkafka). Works standalone (Express, Fastify, raw Node) or as a NestJS DynamicModule. Not a full-featured framework — just a clean, typed layer for producing and consuming Kafka messages.

This library exists so you don't have to think about:

• rebalance edge cases
• retry loops and backoff scheduling
• dead letter queue wiring
• transaction coordinator warmup
• graceful shutdown and offset commit pitfalls
• silent message loss

Safe by default. Configurable when you need it. Escape hatches for when you know what you're doing.

Why?

• Typed topics — you define a map of topic -> message shape, and the compiler won't let you send wrong data to wrong topic
• Topic descriptors — topic() DX sugar lets you define topics as standalone typed objects instead of string keys
• Framework-agnostic — use standalone or with NestJS (register() / registerAsync(), DI, lifecycle hooks)
• Idempotent producer — acks: -1, idempotent: true by default
• Lamport Clock deduplication — every outgoing message is stamped with a monotonically increasing x-lamport-clock header; the consumer tracks the last processed value per topic:partition and silently drops (or routes to DLQ / a dedicated topic) any message whose clock is not strictly greater than the last seen value
• Retry + DLQ — exponential backoff with full jitter; dead letter queue with error metadata headers (original topic, error message, stack, attempt count)
• Batch sending — send multiple messages in a single request
• Batch consuming — startBatchConsumer() for high-throughput eachBatch processing
• Partition key support — route related messages to the same partition
• Custom headers — attach metadata headers to messages
• Transactions — exactly-once semantics with producer.transaction()
• EventEnvelope — every consumed message is wrapped in EventEnvelope<T> with eventId, correlationId, timestamp, schemaVersion, traceparent, and Kafka metadata
• Correlation ID propagation — auto-generated on send, auto-propagated through AsyncLocalStorage so nested sends inherit the same correlation ID
• OpenTelemetry support — @drarzter/kafka-client/otel entrypoint with otelInstrumentation() for W3C Trace Context propagation
• Consumer interceptors — before/after/onError hooks with EventEnvelope access
• Client-wide instrumentation — KafkaInstrumentation hooks for cross-cutting concerns (tracing, metrics)
• Auto-create topics — autoCreateTopics: true for dev mode — no need to pre-create topics
• Error classes — KafkaProcessingError and KafkaRetryExhaustedError with topic, message, and attempt metadata
• Health check — built-in health indicator for monitoring
• Multiple consumer groups — named clients for different bounded contexts
• Declarative & imperative — use @SubscribeTo() decorator or startConsumer() directly
• Async iterator — consume<K>() returns an AsyncIterableIterator<EventEnvelope<T[K]>> for for await consumption; breaking out of the loop stops the consumer automatically
• Message TTL — messageTtlMs drops or DLQs messages older than a configurable threshold, preventing stale events from poisoning downstream systems after a lag spike
• Circuit breaker — circuitBreaker option applies a sliding-window breaker per topic-partition; pauses delivery on repeated handler failures and resumes after a configurable recovery window
• Seek to offset — seekToOffset(groupId, assignments) seeks individual partitions to explicit offsets for fine-grained replay
• Tombstone messages — sendTombstone(topic, key) sends a null-value record to compact a key out of a log-compacted topic; all instrumentation hooks still fire
• Regex topic subscription — startConsumer([/^orders\..+/], handler) subscribes using a pattern; the broker routes matching topics to the consumer dynamically
• Compression — per-send compression option (gzip, snappy, lz4, zstd) in SendOptions and BatchSendOptions
• Partition assignment strategy — partitionAssigner in ConsumerOptions chooses between cooperative-sticky (default), roundrobin, and range
• Admin API — listConsumerGroups(), describeTopics(), deleteRecords() for group inspection, partition metadata, and message deletion
• Typed partition keys — topic('orders').type<T>().key(m => m.orderId) binds a partition-key extractor to a descriptor so related messages land on the same partition without passing key at every call site
• Versioned schemas — versionedSchema({ 1: v1, 2: v2 }, { migrate }) dispatches validation on the x-schema-version header and upgrades old shapes to the latest
• Constructor validation — the KafkaClient constructor fails fast, throwing a single aggregated error that lists every invalid config value instead of surfacing a confusing driver error on first use
• Pluggable deduplication store — swap the in-memory Lamport-clock store for a DedupStore (e.g. Redis-backed) so deduplication survives restarts and rebalances; fail-open on store errors
• Delayed delivery — sendMessage(..., { deliverAfterMs }) stages messages in <topic>.delayed; a startDelayedRelay() consumer forwards them transactionally once the deadline passes
• OpenTelemetry metrics — otelMetricsInstrumentation() records send/consume counters and a handler-duration histogram; otelLagGauge() reports per-partition consumer lag as an observable gauge
• Transport security — security: { ssl, sasl } with secure-by-default rules: SASL auto-enables TLS, plaintext to non-local brokers warns once (silenceable via allowInsecure: true); SASL mechanisms plain, scram-sha-256, scram-sha-512, oauthbearer
• AWS MSK / GCP auth — awsMskIamProvider({ region }) and gcpAccessTokenProvider() supply OAUTHBEARER tokens from the standard AWS / Google credential chains (IRSA, task roles, ADC)
• ACL requirements helper — describeRequiredAcls() enumerates every derived topic, companion group, ephemeral group, and transactional id a service needs; render them as kafka-acls.sh commands or an MSK IAM policy
• Environment configuration — kafkaClientConfigFromEnv(), consumerOptionsFromEnv(), and mergeConsumerOptions() build config from env vars with code > env > defaults precedence
• Transactional outbox — startOutboxRelay() publishes rows from a DB outbox table to Kafka inside a transaction; at-least-once with stable eventId for downstream dedup
• Pluggable serialization — JSON by default; drop in avroSerde() / protobufSerde() (@drarzter/kafka-client/serde) for Confluent wire-format Avro/Protobuf and interop with Java/Go via a Schema Registry, client-wide or per-topic
• Schema Registry client — SchemaRegistryClient + registrySchema() keep locally-defined schemas in lockstep with a Confluent-compatible registry
• Static group membership — groupInstanceId (group.instance.id) skips rebalance on k8s rolling restarts within session.timeout.ms
• DLQ CLI — kafka-client-dlq ls | peek | replay for inspecting and re-publishing dead letter queues from the terminal

See the Roadmap for upcoming features and version history.

How it compares

Kafka in Node.js is usually one of: the low-level driver kafkajs (and things built on it, like the NestJS Kafka transport), or the native @confluentinc/kafka-javascript driver. Those give you a solid transport — produce, consume, commit — and leave the reliability patterns (retry topologies, DLQ, circuit breaking, dedup, delayed delivery, serde) for you to build and maintain yourself. This library is that reliability layer, batteries-included, on top of the Confluent/librdkafka driver.

The table is about what ships out of the box — not what's theoretically buildable. kafkajs is a driver, so most rows below are "do it yourself" there; that's the point.

| Capability | @drarzter/kafka-client | kafkajs | @nestjs/microservices (Kafka) | |---|:---:|:---:|:---:| | Compile-time typed topic → payload map | ✅ | ❌ | ❌ | | Produce / consume / batch / admin | ✅ | ✅ | ✅ | | Exactly-once transactions | ✅ | ✅ | ⚠️ limited | | Retry with backoff + jitter | ✅ built-in | 🔨 DIY | 🔨 DIY | | Durable retry-topic chains (<topic>.retry.N) | ✅ | 🔨 DIY | 🔨 DIY | | Dead-letter queue + metadata headers | ✅ | 🔨 DIY | 🔨 DIY | | Circuit breaker (per partition) | ✅ | 🔨 DIY | 🔨 DIY | | Deduplication (Lamport clock, pluggable store) | ✅ | 🔨 DIY | 🔨 DIY | | Delayed delivery + transactional relay | ✅ | 🔨 DIY | 🔨 DIY | | Transactional outbox relay | ✅ | 🔨 DIY | 🔨 DIY | | Avro / Protobuf serde (Confluent wire format) | ✅ /serde | 🔨 DIY | ❌ | | OpenTelemetry traces and metrics | ✅ | 🔨 DIY | 🔨 DIY | | Envelope (eventId / correlationId / trace) + ALS propagation | ✅ | ❌ | ❌ | | Security helpers (MSK IAM / GCP OAUTHBEARER, ACL generator) | ✅ | ⚠️ manual | ⚠️ manual | | First-class NestJS integration | ✅ | ❌ | ✅ |

✅ built-in · ⚠️ partial / manual config · 🔨 possible but you build & maintain it · ❌ not available

@nestjs/microservices Kafka transport is itself built on kafkajs and targets request-response / event messaging patterns rather than data-streaming reliability — so it inherits the same "build it yourself" gaps.

And it's nearly free. The throughput benchmark (npm run bench) measures this wrapper against the raw @confluentinc/kafka-javascript driver on a real broker: ~2% overhead with identical p50/p95 latency. The typed envelope, Lamport clock, and instrumentation hooks cost almost nothing on the hot path — you get every row above without trading away throughput.

The comparison reflects built-in capabilities as of this version and isn't a knock on the driver-level libraries — kafkajs is a fine transport; this just saves you from re-implementing the layer above it.

Installation

npm install @drarzter/kafka-client

@confluentinc/kafka-javascript uses a native librdkafka addon. On most systems it builds automatically. For faster installs (skips compilation), install the system library first:

# Arch / CachyOS
sudo pacman -S librdkafka

# Debian / Ubuntu
sudo apt-get install librdkafka-dev

# macOS
brew install librdkafka

Then install with BUILD_LIBRDKAFKA=0 npm install.

For NestJS projects, install peer dependencies: @nestjs/common, @nestjs/core, reflect-metadata, rxjs.

For standalone usage (Express, Fastify, raw Node), no extra dependencies needed — import from @drarzter/kafka-client/core.

Standalone usage (no NestJS)

import { KafkaClient, topic } from '@drarzter/kafka-client/core';

const OrderCreated = topic('order.created').type<{ orderId: string; amount: number }>();

const kafka = new KafkaClient('my-app', 'my-group', ['localhost:9092']);
await kafka.connectProducer();

// Send
await kafka.sendMessage(OrderCreated, { orderId: '123', amount: 100 });

// Consume — handler receives an EventEnvelope
await kafka.startConsumer([OrderCreated], async (envelope) => {
  console.log(`${envelope.topic}:`, envelope.payload.orderId);
});

// Custom logger (winston, pino, etc.)
const kafka2 = new KafkaClient('my-app', 'my-group', ['localhost:9092'], {
  logger: myWinstonLogger,
});

// All module options work in standalone mode too
const kafka3 = new KafkaClient('my-app', 'my-group', ['localhost:9092'], {
  autoCreateTopics: true,   // auto-create topics on first use
  numPartitions: 3,         // partitions for auto-created topics
  strictSchemas: false,     // disable schema enforcement for string topic keys
  instrumentation: [...],   // client-wide tracing/metrics hooks
});

// Health check — available directly, no NestJS needed
const status = await kafka.checkStatus();
// { status: 'up', clientId: 'my-app', topics: ['order.created', ...] }

// Stop all consumers without disconnecting the producer or admin
// Useful when you want to re-subscribe with different options
await kafka.stopConsumer();

Quick start (NestJS)

Send and receive a message in 3 files:

// types.ts
export interface MyTopics {
  'hello': { text: string };
}

// app.module.ts
import { Module } from '@nestjs/common';
import { KafkaModule } from '@drarzter/kafka-client';
import { MyTopics } from './types';
import { AppService } from './app.service';

@Module({
  imports: [
    KafkaModule.register<MyTopics>({
      clientId: 'my-app',
      groupId: 'my-group',
      brokers: ['localhost:9092'],
    }),
  ],
  providers: [AppService],
})
export class AppModule {}

// app.service.ts
import { Injectable } from '@nestjs/common';
import { InjectKafkaClient, KafkaClient, SubscribeTo, EventEnvelope } from '@drarzter/kafka-client';
import { MyTopics } from './types';

@Injectable()
export class AppService {
  constructor(
    @InjectKafkaClient() private readonly kafka: KafkaClient<MyTopics>,
  ) {}

  async send() {
    await this.kafka.sendMessage('hello', { text: 'Hello, Kafka!' });
  }

  @SubscribeTo('hello')
  async onHello(envelope: EventEnvelope<MyTopics['hello']>) {
    console.log('Received:', envelope.payload.text);
  }
}

Usage

1. Define your topic map

Both interface and type work — pick whichever you prefer:

// Explicit: extends TTopicMessageMap — IDE hints that values must be Record<string, any>
import { TTopicMessageMap } from '@drarzter/kafka-client';

export interface OrdersTopicMap extends TTopicMessageMap {
  'order.created': {
    orderId: string;
    userId: string;
    amount: number;
  };
  'order.completed': {
    orderId: string;
    completedAt: string;
  };
}

// Minimal: plain interface or type — works just the same
export interface OrdersTopicMap {
  'order.created': { orderId: string; userId: string; amount: number };
  'order.completed': { orderId: string; completedAt: string };
}

// or
export type OrdersTopicMap = {
  'order.created': { orderId: string; userId: string; amount: number };
  'order.completed': { orderId: string; completedAt: string };
};

Alternative: topic() descriptors

Instead of a centralized topic map, define each topic as a standalone typed object:

import { topic, TopicsFrom } from '@drarzter/kafka-client';

export const OrderCreated = topic('order.created').type<{
  orderId: string;
  userId: string;
  amount: number;
}>();

export const OrderCompleted = topic('order.completed').type<{
  orderId: string;
  completedAt: string;
}>();

// Combine into a topic map for KafkaModule generics
export type OrdersTopicMap = TopicsFrom<typeof OrderCreated | typeof OrderCompleted>;

Topic descriptors work everywhere strings work — sendMessage, sendBatch, transaction, startConsumer, and @SubscribeTo():

// Sending
await kafka.sendMessage(OrderCreated, { orderId: '123', userId: '456', amount: 100 });
await kafka.sendBatch(OrderCreated, [{ value: { orderId: '1', userId: '10', amount: 50 } }]);

// Transactions
await kafka.transaction(async (tx) => {
  await tx.send(OrderCreated, { orderId: '123', userId: '456', amount: 100 });
});

// Consuming (decorator)
@SubscribeTo(OrderCreated)
async handleOrder(envelope: EventEnvelope<OrdersTopicMap['order.created']>) { ... }

// Consuming (imperative)
await kafka.startConsumer([OrderCreated], handler);

2. Register the module

import { KafkaModule } from '@drarzter/kafka-client';
import { OrdersTopicMap } from './orders.types';

@Module({
  imports: [
    KafkaModule.register<OrdersTopicMap>({
      clientId: 'my-service',
      groupId: 'my-consumer-group',
      brokers: ['localhost:9092'],
      autoCreateTopics: true, // auto-create topics on first use (dev mode)
    }),
  ],
})
export class OrdersModule {}

autoCreateTopics calls admin.createTopics() (idempotent — no-op if topic already exists) before the first send and before each startConsumer / startBatchConsumer call. librdkafka errors on unknown topics at subscribe time, so consumer-side creation is required. Useful in development, not recommended for production.

Or with ConfigService:

KafkaModule.registerAsync<OrdersTopicMap>({
  imports: [ConfigModule],
  inject: [ConfigService],
  useFactory: (config: ConfigService) => ({
    clientId: 'my-service',
    groupId: 'my-consumer-group',
    brokers: config.get<string>('KAFKA_BROKERS').split(','),
  }),
})

Global module

By default, KafkaModule is scoped — you need to import it in every module that uses @InjectKafkaClient(). Pass isGlobal: true to make the client available everywhere:

// app.module.ts — register once
KafkaModule.register<OrdersTopicMap>({
  clientId: 'my-service',
  groupId: 'my-consumer-group',
  brokers: ['localhost:9092'],
  isGlobal: true,
})

// any other module — no need to import KafkaModule
@Injectable()
export class SomeService {
  constructor(@InjectKafkaClient() private readonly kafka: KafkaClient<OrdersTopicMap>) {}
}

Works with registerAsync() too:

KafkaModule.registerAsync<OrdersTopicMap>({
  isGlobal: true,
  imports: [ConfigModule],
  inject: [ConfigService],
  useFactory: (config: ConfigService) => ({ ... }),
})

3. Inject and use

import { Injectable } from '@nestjs/common';
import { InjectKafkaClient, KafkaClient } from '@drarzter/kafka-client';
import { OrdersTopicMap } from './orders.types';

@Injectable()
export class OrdersService {
  constructor(
    @InjectKafkaClient()
    private readonly kafka: KafkaClient<OrdersTopicMap>,
  ) {}

  async createOrder() {
    await this.kafka.sendMessage('order.created', {
      orderId: '123',
      userId: '456',
      amount: 100,
    });
  }
}

Consuming messages

Three ways — choose what fits your style.

Declarative: @SubscribeTo()

import { Injectable } from '@nestjs/common';
import { SubscribeTo } from '@drarzter/kafka-client';

@Injectable()
export class OrdersHandler {
  @SubscribeTo('order.created')
  async handleOrderCreated(envelope: EventEnvelope<OrdersTopicMap['order.created']>) {
    console.log('New order:', envelope.payload.orderId);
  }

  @SubscribeTo('order.completed', { retry: { maxRetries: 3 }, dlq: true })
  async handleOrderCompleted(envelope: EventEnvelope<OrdersTopicMap['order.completed']>) {
    console.log('Order completed:', envelope.payload.orderId);
  }
}

The module auto-discovers @SubscribeTo() methods on startup and subscribes them.

Imperative: startConsumer()

@Injectable()
export class OrdersService implements OnModuleInit {
  constructor(
    @InjectKafkaClient()
    private readonly kafka: KafkaClient<OrdersTopicMap>,
  ) {}

  async onModuleInit() {
    await this.kafka.startConsumer(
      ['order.created', 'order.completed'],
      async (envelope) => {
        console.log(`${envelope.topic}:`, envelope.payload);
      },
      {
        retry: { maxRetries: 3, backoffMs: 1000 },
        dlq: true,
      },
    );
  }
}

Regex topic subscription

Subscribe to multiple topics matching a pattern — the broker dynamically routes any topic whose name matches the regex to this consumer:

// Subscribe to all topics starting with "orders."
await kafka.startConsumer([/^orders\..+/], handler);

// Mix regexes and literal strings
await kafka.startConsumer([/^payments\..+/, 'audit.global'], handler);

Works with startBatchConsumer and @SubscribeTo too:

@SubscribeTo(/^events\..+/)
async handleEvent(envelope: EventEnvelope<any>) { ... }

Limitation: retryTopics: true is incompatible with regex subscriptions — the library cannot derive static retry topic names from a pattern. An error is thrown at startup if both are combined.

Iterator: consume()

Stream messages from a single topic as an AsyncIterableIterator — useful for scripts, one-off tasks, or any context where you prefer for await over a callback:

for await (const envelope of kafka.consume('order.created')) {
  console.log('Order:', envelope.payload.orderId);
}

// Breaking out of the loop stops the consumer automatically
for await (const envelope of kafka.consume('order.created')) {
  if (envelope.payload.orderId === targetId) break;
}

consume() accepts the same ConsumerOptions as startConsumer():

for await (const envelope of kafka.consume('orders', {
  retry: { maxRetries: 3 },
  dlq: true,
  messageTtlMs: 60_000,
})) {
  await processOrder(envelope.payload);
}

break, return, or any early exit from the loop calls the iterator's return() method, which closes the internal queue and calls handle.stop() on the background consumer.

Backpressure — use queueHighWaterMark to prevent unbounded queue growth when processing is slower than the message rate:

for await (const envelope of kafka.consume('orders', {
  queueHighWaterMark: 100, // pause partition when queue reaches 100 messages
})) {
  await slowProcessing(envelope.payload); // resumes when queue drains below 50
}

The partition is paused when the internal queue reaches queueHighWaterMark and automatically resumed when it drains below 50%. Without this option the queue is unbounded.

Error propagation — if the consumer fails to start (e.g. broker unreachable), the error surfaces on the next next() / for await iteration rather than being silently swallowed.

Multiple consumer groups

Per-consumer groupId

Override the default consumer group for specific consumers. Each unique groupId creates a separate librdkafka Consumer internally:

// Default group from constructor
await kafka.startConsumer(['orders'], handler);

// Custom group — receives its own copy of messages
await kafka.startConsumer(['orders'], auditHandler, { groupId: 'orders-audit' });

// Works with @SubscribeTo too
@SubscribeTo('orders', { groupId: 'orders-audit' })
async auditOrders(envelope) { ... }

Important: You cannot mix eachMessage and eachBatch consumers on the same groupId, and you cannot call startConsumer (or startBatchConsumer) twice on the same groupId without stopping it first. The library throws a clear error in both cases:

Cannot use eachBatch on consumer group "my-group" — it is already running with eachMessage.
Use a different groupId for this consumer.

startConsumer("my-group") called twice — this group is already consuming.
Call stopConsumer("my-group") first or pass a different groupId.

Named clients

Register multiple named clients for different bounded contexts:

@Module({
  imports: [
    KafkaModule.register<OrdersTopicMap>({
      name: 'orders',
      clientId: 'orders-service',
      groupId: 'orders-consumer',
      brokers: ['localhost:9092'],
    }),
    KafkaModule.register<PaymentsTopicMap>({
      name: 'payments',
      clientId: 'payments-service',
      groupId: 'payments-consumer',
      brokers: ['localhost:9092'],
    }),
  ],
})
export class AppModule {}

Inject by name — the string in @InjectKafkaClient() must match the name from register():

@Injectable()
export class OrdersService {
  constructor(
    @InjectKafkaClient('orders')    // ← matches name: 'orders' above
    private readonly kafka: KafkaClient<OrdersTopicMap>,
  ) {}
}

Same with @SubscribeTo() — use clientName to target a specific named client:

@SubscribeTo('payment.received', { clientName: 'payments' })  // ← matches name: 'payments'
async handlePayment(envelope: EventEnvelope<PaymentsTopicMap['payment.received']>) {
  // ...
}

Partition key

Route all events for the same order to the same partition:

await this.kafka.sendMessage(
  'order.created',
  { orderId: '123', userId: '456', amount: 100 },
  { key: '123' },
);

Typed partition keys

Instead of passing key at every call site, bind a partition-key extractor to the topic descriptor with .key(). The extractor runs on every send through that descriptor, so messages with the same logical key always land on the same partition — you never forget to set it. Available on both .type<T>() and .schema() descriptors:

import { topic } from '@drarzter/kafka-client';

const OrderCreated = topic('order.created')
  .type<{ orderId: string; userId: string; amount: number }>()
  .key((m) => m.orderId);

// Key is derived automatically from the payload — no `key` needed
await kafka.sendMessage(OrderCreated, { orderId: '123', userId: '456', amount: 100 });
// → produced with key '123'

// Works with schema descriptors too
const PaymentTaken = topic('payment.taken')
  .schema(z.object({ paymentId: z.string(), orderId: z.string() }))
  .key((m) => m.orderId);

The extractor runs on the original (pre-validation) payload. An explicit key in SendOptions — or a batch item's key — always wins over the descriptor's extractor:

// Explicit key overrides the extractor
await kafka.sendMessage(OrderCreated, { orderId: '123', userId: '456', amount: 100 }, {
  key: 'custom-partition-key',
});

Message headers

Attach metadata to messages:

await this.kafka.sendMessage(
  'order.created',
  { orderId: '123', userId: '456', amount: 100 },
  {
    key: '123',
    headers: { 'x-correlation-id': 'abc-def', 'x-source': 'api-gateway' },
  },
);

Headers work with batch sending too:

await this.kafka.sendBatch('order.created', [
  {
    value: { orderId: '1', userId: '10', amount: 50 },
    key: '1',
    headers: { 'x-correlation-id': 'req-1' },
  },
]);

Batch sending

await this.kafka.sendBatch('order.created', [
  { value: { orderId: '1', userId: '10', amount: 50 }, key: '1' },
  { value: { orderId: '2', userId: '20', amount: 75 }, key: '2' },
  { value: { orderId: '3', userId: '30', amount: 100 }, key: '3' },
]);

Delayed delivery

Schedule a message for future delivery with deliverAfterMs. Instead of going straight to the target topic, the message is produced to a <topic>.delayed staging topic carrying x-delayed-until (deadline) and x-delayed-target headers. A relay consumer started via startDelayedRelay() holds each message until its deadline passes, then forwards it to the target topic:

// 1. Start the relay once (per process) for the topics you delay-deliver to
await kafka.startDelayedRelay(['order.reminder']);

// 2. Send a message that should arrive in ~1 hour
await kafka.sendMessage(
  'order.reminder',
  { orderId: '123', channel: 'email' },
  { deliverAfterMs: 60 * 60 * 1000 },
);
// → staged in order.reminder.delayed, forwarded to order.reminder ~1 h later

deliverAfterMs also works on sendBatch — it applies to the whole batch:

await kafka.sendBatch('order.reminder', messages, { deliverAfterMs: 30_000 });

The relay defaults to a <defaultGroupId>-delayed-relay consumer group; override it with startDelayedRelay(topics, { groupId }). Forwarding is transactional — the produce to the target topic and the source-offset commit happen atomically, so no duplicates are relayed even if the relay crashes mid-forward. The original key, value, and envelope headers (x-event-id, x-correlation-id, x-lamport-clock, traceparent) all survive the hop; only the x-delayed-* control headers are stripped.

Delivery time is a lower bound. The relay pauses a partition until the head-of-line message's deadline, so later messages on the same partition wait behind it (at-least semantics). Delayed messages are only delivered while the relay is running — treat it as a long-lived consumer, not a fire-and-forget scheduler.

Batch consuming

Process messages in batches for higher throughput. The handler receives an array of EventEnvelopes and a BatchMeta object with offset management controls:

await this.kafka.startBatchConsumer(
  ['order.created'],
  async (envelopes, meta) => {
    // envelopes: EventEnvelope<OrdersTopicMap['order.created']>[]
    for (const env of envelopes) {
      await processOrder(env.payload);
      meta.resolveOffset(env.offset);

      // Call heartbeat() during long-running batch processing to prevent
      // the broker from considering the consumer dead (session.timeout.ms)
      await meta.heartbeat();
    }
    await meta.commitOffsetsIfNecessary();
  },
  { retry: { maxRetries: 3 }, dlq: true },
);

With autoCommit: false for full manual offset control:

await this.kafka.startBatchConsumer(
  ['order.created'],
  async (envelopes, meta) => {
    for (const env of envelopes) {
      await processOrder(env.payload);
      meta.resolveOffset(env.offset);
    }
    // commitOffsetsIfNecessary() commits only when autoCommit is off
    // or when the commit interval has elapsed
    await meta.commitOffsetsIfNecessary();
  },
  { autoCommit: false },
);

Note: If your handler calls resolveOffset() or commitOffsetsIfNecessary() without setting autoCommit: false, a debug message is logged at consumer-start time — mixing autoCommit with manual offset control causes offset conflicts. Set autoCommit: false to suppress the message and take full control of offset management.

With @SubscribeTo():

@SubscribeTo('order.created', { batch: true })
async handleOrders(envelopes: EventEnvelope<OrdersTopicMap['order.created']>[], meta: BatchMeta) {
  for (const env of envelopes) { ... }
}

Schema validation runs per-message — invalid messages are skipped (DLQ'd if enabled), valid ones are passed to the handler. Retry applies to the whole batch.

retryTopics: true is also supported on startBatchConsumer. On handler failure, each envelope in the batch is routed individually to <topic>.retry.1; the companion retry consumers call the batch handler one message at a time with a stub BatchMeta (no-op heartbeat/resolveOffset/commitOffsetsIfNecessary):

await kafka.startBatchConsumer(
  ['orders.created'],
  async (envelopes, meta) => { /* same handler */ },
  {
    retry: { maxRetries: 3, backoffMs: 1000 },
    dlq: true,
    retryTopics: true, // ← now supported for batch consumers too
  },
);

BatchMeta exposes:

| Property/Method | Description | | --------------- | ----------- | | partition | Partition number for this batch | | highWatermark | Latest offset in the partition (string). null when the message is replayed via a retry topic consumer — in that path the broker high-watermark is not available. Guard against null before computing lag | | heartbeat() | Send a heartbeat to keep the consumer session alive — call during long processing loops | | resolveOffset(offset) | Mark offset as processed (required before commitOffsetsIfNecessary) | | commitOffsetsIfNecessary() | Commit resolved offsets; respects autoCommit setting |

Tombstone messages

Send a null-value Kafka record to compact a specific key out of a log-compacted topic:

// Delete the record with key "user-123" from the log-compacted "users" topic
await kafka.sendTombstone('users', 'user-123');

// With custom headers
await kafka.sendTombstone('users', 'user-123', { 'x-reason': 'gdpr-deletion' });

sendTombstone skips envelope headers, schema validation, and the Lamport clock — the record value is literally null, as required by Kafka's log compaction protocol. Both beforeSend and afterSend instrumentation hooks still fire so tracing works correctly.

Compression

Reduce network bandwidth with per-send compression. Supported codecs: 'gzip', 'snappy', 'lz4', 'zstd':

import { CompressionType } from '@drarzter/kafka-client/core';

// Single message
await kafka.sendMessage('events', payload, { compression: 'gzip' });

// Batch
await kafka.sendBatch('events', messages, { compression: 'snappy' });

Compression is applied at the Kafka message-set level — the broker decompresses transparently on the consumer side. 'snappy' and 'lz4' offer the best throughput/CPU trade-off for most workloads; 'gzip' gives the highest compression ratio; 'zstd' balances both.

Transactions

Send multiple messages atomically with exactly-once semantics:

await this.kafka.transaction(async (tx) => {
  await tx.send('order.created', {
    orderId: '123',
    userId: '456',
    amount: 100,
  });
  await tx.send('order.completed', {
    orderId: '123',
    completedAt: new Date().toISOString(),
  });
  // if anything throws, all messages are rolled back
});

tx.sendBatch() is also available inside transactions:

await this.kafka.transaction(async (tx) => {
  await tx.sendBatch('order.created', [
    { value: { orderId: '1', userId: '10', amount: 50 }, key: '1' },
    { value: { orderId: '2', userId: '20', amount: 75 }, key: '2' },
  ]);
  // if anything throws, all messages are rolled back
});

Consumer interceptors

Add before/after/onError hooks to message processing. Interceptors receive the full EventEnvelope:

import { ConsumerInterceptor } from '@drarzter/kafka-client';

const loggingInterceptor: ConsumerInterceptor<OrdersTopicMap> = {
  before: (envelope) => {
    console.log(`Processing ${envelope.topic}`, envelope.payload);
  },
  after: (envelope) => {
    console.log(`Done ${envelope.topic}`);
  },
  onError: (envelope, error) => {
    console.error(`Failed ${envelope.topic}:`, error.message);
  },
};

await this.kafka.startConsumer(['order.created'], handler, {
  interceptors: [loggingInterceptor],
});

Multiple interceptors run in order. All hooks are optional.

Instrumentation

For client-wide cross-cutting concerns (tracing, metrics), use KafkaInstrumentation hooks instead of per-consumer interceptors:

import { otelInstrumentation } from '@drarzter/kafka-client/otel';

const kafka = new KafkaClient('my-app', 'my-group', brokers, {
  instrumentation: [otelInstrumentation()],
});

otelInstrumentation() injects traceparent on send, extracts it on consume, and creates CONSUMER spans automatically. The span is set as the active OTel context for the handler's duration via context.with() — so trace.getActiveSpan() works inside your handler and any child spans are automatically parented to the consume span. Requires @opentelemetry/api as a peer dependency.

OpenTelemetry metrics

otelInstrumentation() handles traces. For metrics, the same entrypoint exports otelMetricsInstrumentation() (counters + a duration histogram) and otelLagGauge() (an observable consumer-lag gauge). They share nothing with the tracing instrumentation and compose with it in any order:

import {
  otelInstrumentation,
  otelMetricsInstrumentation,
  otelLagGauge,
} from '@drarzter/kafka-client/otel';

const kafka = new KafkaClient('my-app', 'my-group', brokers, {
  instrumentation: [otelInstrumentation(), otelMetricsInstrumentation()],
});

otelMetricsInstrumentation() registers seven instruments under the meter @drarzter/kafka-client (created once per instance, not per message):

| Instrument | Type | Attributes | Recorded when | | ---------- | ---- | ---------- | ------------- | | kafka.client.messages.sent | Counter | topic | a message is sent | | kafka.client.messages.processed | Counter | topic | a handler succeeds | | kafka.client.messages.retried | Counter | topic | a message is queued for retry | | kafka.client.messages.dlq | Counter | topic, reason | a message is routed to a DLQ | | kafka.client.messages.duplicate | Counter | topic, strategy | a Lamport-clock duplicate is detected | | kafka.client.consume.errors | Counter | topic | a handler throws | | kafka.client.consume.duration | Histogram (ms) | topic | measured across the handler's execution |

Pass a custom meter with otelMetricsInstrumentation({ meter }) to route instruments through your own MeterProvider; it defaults to metrics.getMeter('@drarzter/kafka-client').

otelLagGauge() registers an observable gauge kafka.client.consumer.lag (attributes topic, partition, groupId) that polls getConsumerLag() on each metric-collection cycle. It returns an unregister disposer — call it on shutdown to stop observing:

const unregisterLag = otelLagGauge(kafka, { groupId: 'billing-service' });

// ...later, on shutdown:
unregisterLag();

groupId defaults to the client's constructor group (reported as an empty-string attribute), and meter overrides the meter as above. Lag-query failures during a collection cycle are swallowed silently — a broker hiccup reports no samples for that cycle rather than breaking metric collection. Both helpers require @opentelemetry/api as a peer dependency.

Custom instrumentation

beforeConsume can return a BeforeConsumeResult — either the legacy () => void cleanup function, or an object with cleanup and/or wrap:

import { KafkaInstrumentation, BeforeConsumeResult } from '@drarzter/kafka-client';

const myInstrumentation: KafkaInstrumentation = {
  beforeSend(topic, headers) { /* inject headers, start timer */ },
  afterSend(topic) { /* record send latency */ },

  beforeConsume(envelope): BeforeConsumeResult {
    const span = startMySpan(envelope.topic);
    return {
      // cleanup() is called after the handler completes (success or error)
      cleanup() { span.end(); },
      // wrap(fn) runs the handler inside the desired async context
      // call fn() wherever you need it in the context scope
      wrap(fn) { return runWithSpanActive(span, fn); },
    };
  },

  onConsumeError(envelope, error) { /* record error metric */ },
};

The legacy () => void form is still fully supported — return a function directly if you only need cleanup:

beforeConsume(envelope) {
  const timer = startTimer();
  return () => timer.end(); // cleanup only, no context wrapping
},

BeforeConsumeResult is a union:

type BeforeConsumeResult =
  | (() => void)                     // legacy: cleanup only
  | { cleanup?(): void;              // called after handler (success or error)
      wrap?(fn: () => Promise<void>): Promise<void>; // wraps handler execution
    };

When multiple instrumentations each provide a wrap, they compose in declaration order — the first instrumentation's wrap is the outermost.

Lifecycle event hooks

Three additional hooks fire for specific events in the consume pipeline:

| Hook | When called | Arguments | | ---- | ----------- | --------- | | onMessage | Handler successfully processed a message | (envelope) — use as a success counter for error-rate calculations | | onRetry | A message is queued for another attempt (in-process backoff or routed to a retry topic) | (envelope, attempt, maxRetries) | | onDlq | A message is routed to the dead letter queue | (envelope, reason) — reason is 'handler-error', 'validation-error', or 'lamport-clock-duplicate' | | onDuplicate | A duplicate is detected via Lamport Clock | (envelope, strategy) — strategy is 'drop', 'dlq', or 'topic' |

const myInstrumentation: KafkaInstrumentation = {
  onMessage(envelope) {
    metrics.increment('kafka.processed', { topic: envelope.topic });
  },
  onRetry(envelope, attempt, maxRetries) {
    console.warn(`Retrying ${envelope.topic} — attempt ${attempt}/${maxRetries}`);
  },
  onDlq(envelope, reason) {
    alertingSystem.send({ topic: envelope.topic, reason });
  },
  onDuplicate(envelope, strategy) {
    metrics.increment('kafka.duplicate', { topic: envelope.topic, strategy });
  },
};

Built-in metrics

KafkaClient maintains lightweight in-process event counters independently of any instrumentation:

// Global snapshot — aggregate across all topics
const snapshot = kafka.getMetrics();
// { processedCount: number; retryCount: number; dlqCount: number; dedupCount: number }

// Per-topic snapshot
const orderMetrics = kafka.getMetrics('order.created');
// { processedCount: 5, retryCount: 1, dlqCount: 0, dedupCount: 0 }

kafka.resetMetrics();                // reset all counters
kafka.resetMetrics('order.created'); // reset only one topic's counters

Passing a topic name that has not seen any events returns a zero-valued snapshot — it never throws.

Counters are incremented in the same code paths that fire the corresponding hooks — they are always active regardless of whether any instrumentation is configured.

Transport security

Configure TLS and SASL through the security option on KafkaClientOptions. The library applies secure-by-default rules so credentials never leak onto plaintext connections by accident:

• SASL auto-enables TLS. When sasl is set and ssl is left unset, ssl is turned on automatically — SASL credentials always travel over TLS unless you explicitly opt out.
• Explicit ssl: false with SASL warns. Setting sasl together with ssl: false logs a warning that credentials will cross the wire in plaintext — only safe on fully trusted networks.
• Plaintext to non-local brokers warns once. With no ssl/sasl at all and at least one non-local broker (anything outside localhost, 127.0.0.0/8, ::1, 0.0.0.0, host.docker.internal), a single warning is logged per client. Acknowledge and silence it with allowInsecure: true.

Nothing here ever throws or blocks a connection — the defaults protect, you stay in control.

import { KafkaClient } from '@drarzter/kafka-client/core';

// SASL/SCRAM over TLS — ssl auto-enabled because sasl is set
const kafka = new KafkaClient('billing-svc', 'billing-group', ['broker.example.com:9093'], {
  security: {
    sasl: {
      mechanism: 'scram-sha-512',
      username: 'billing-svc',
      password: process.env.KAFKA_PASSWORD!,
    },
    // ssl: true — inferred automatically; set explicitly if you prefer
  },
});

KafkaSecurityOptions:

| Field | Default | Description | | ----- | ------- | ----------- | | ssl | true when sasl set, else false | Enable TLS | | sasl | — | SASL authentication (see below) | | allowInsecure | false | Acknowledge an intentionally insecure (plaintext, non-local) setup and silence the warning. No effect when ssl/sasl are set |

sasl is a discriminated union on mechanism:

// Username / password mechanisms
{ mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512', username: string, password: string }

// Token-based (AWS MSK IAM, GCP, custom)
{ mechanism: 'oauthbearer', oauthBearerProvider: () => Promise<OAuthBearerToken> }

An OAuthBearerProvider is an async factory the driver calls on connect and before each token expiry; it returns { value, principal?, lifetimeMs?, extensions? }.

AWS MSK IAM & GCP authentication

Two ready-made oauthbearer providers cover the common managed-Kafka cases. Both resolve credentials from the platform's standard chain — nothing to hard-code — and rely on an optional peer dependency you install alongside this library.

AWS MSK IAM — awsMskIamProvider({ region }) delegates token signing to aws-msk-iam-sasl-signer-js. Credentials come from the standard AWS provider chain, so EKS IRSA, ECS task roles, and env credentials all work unchanged. Authorisation is then governed by IAM policies (kafka-cluster:* actions) — see ACL requirements to generate one:

npm install aws-msk-iam-sasl-signer-js

import { KafkaClient, awsMskIamProvider } from '@drarzter/kafka-client/core';

const kafka = new KafkaClient('orders-svc', 'orders-group', brokers, {
  security: {
    sasl: {
      mechanism: 'oauthbearer',
      oauthBearerProvider: awsMskIamProvider({ region: 'eu-west-1' }),
    },
  },
});

GCP — gcpAccessTokenProvider() delegates to google-auth-library using Application Default Credentials, so GKE Workload Identity, attached service accounts, and GOOGLE_APPLICATION_CREDENTIALS all work unchanged. It supplies a raw ADC access token; verify the exact token format your cluster expects against current Google documentation:

npm install google-auth-library

import { KafkaClient, gcpAccessTokenProvider } from '@drarzter/kafka-client/core';

const kafka = new KafkaClient('events-svc', 'events-group', brokers, {
  security: {
    sasl: {
      mechanism: 'oauthbearer',
      oauthBearerProvider: gcpAccessTokenProvider(),
    },
  },
});

| Provider | Options | Optional peer dep | | -------- | ------- | ----------------- | | awsMskIamProvider | { region } | aws-msk-iam-sasl-signer-js | | gcpAccessTokenProvider | { scopes?, principal?, tokenTtlMs? } (defaults: cloud-platform scope, principal 'gcp', 50 min TTL) | google-auth-library |

Neither package is a hard dependency — they are dynamically imported on first token fetch. If the package is missing, the provider throws a clear install hint rather than failing at build time.

ACL requirements

The features that make this library convenient — retry topics, DLQ, delayed delivery, deduplication routing, DLQ replay, snapshots, clock recovery — quietly create extra topics and consumer groups (<topic>.retry.N, <topic>.dlq, <topic>.delayed, <topic>.duplicates, <groupId>-retry.N, timestamped ephemeral groups, transactional ids). On a locked-down cluster every one of them needs an ACL, and the last place you want to discover a missing grant is production at 3 a.m.

describeRequiredAcls() enumerates the complete set from a declarative usage profile. Feed the result to toKafkaAclCommands() for kafka-acls.sh commands, or toMskIamPolicy() for an AWS MSK IAM policy document:

import {
  describeRequiredAcls,
  toKafkaAclCommands,
  toMskIamPolicy,
} from '@drarzter/kafka-client/core';

const resources = describeRequiredAcls({
  clientId: 'billing-svc',
  groupIds: ['billing-svc-group'],
  produceTopics: ['invoices.created'],
  consumeTopics: ['orders.created'],
  features: {
    retryTopics: { maxRetries: 3 },
    dlq: true,
    dlqReplay: true,
    transactions: true,
  },
});

// Render kafka-acls.sh commands for a principal
for (const cmd of toKafkaAclCommands(resources, 'User:billing-svc', 'broker:9092')) {
  console.log(cmd);
}
// kafka-acls.sh --bootstrap-server broker:9092 --add --allow-principal 'User:billing-svc' \
//   --operation READ --operation DESCRIBE --topic 'orders.created'  # startConsumer
// kafka-acls.sh ... --topic 'orders.created.dlq'   # dlq: true — failed messages routed to DLQ
// kafka-acls.sh ... --topic 'orders.created.retry.1' ... --topic 'orders.created.retry.3'
// kafka-acls.sh ... --group 'billing-svc-group-retry.' --resource-pattern-type prefixed
// kafka-acls.sh ... --transactional-id 'billing-svc-group-' --resource-pattern-type prefixed
// kafka-acls.sh ... --group 'orders.created.dlq-replay' --operation DELETE --resource-pattern-type prefixed
// ...

// Or an MSK IAM policy document
const policy = toMskIamPolicy(resources, {
  region: 'eu-west-1',
  accountId: '123456789012',
  clusterName: 'prod',
  clusterUuid: 'abcd-1234',
});

describeRequiredAcls() returns AclResource[], each carrying resourceType (topic | group | transactional-id | cluster), patternType (literal | prefixed), name, operations, and a reason naming the feature that requires it. Ephemeral-group features (dlqReplay, snapshots, clockRecovery) request DELETE on a prefixed pattern, because those groups are timestamped and cleaned up after use.

| Feature flag | Adds | | ------------ | ---- | | dlq | <topic>.dlq WRITE per consumed topic | | retryTopics: { maxRetries } | <topic>.retry.1…N topics; <groupId>-retry. prefixed groups; <groupId>- prefixed transactional ids | | delayedDelivery | <topic>.delayed topics; <groupId>-delayed-relay group + -tx id | | duplicatesTopic | <topic>.duplicates (or a custom topic name) WRITE | | dlqReplay | <topic>.dlq-replay prefixed groups (READ, DESCRIBE, DELETE) + DLQ READ | | snapshots | <clientId>-snapshot- prefixed groups (READ, DESCRIBE, DELETE) | | clockRecovery | <clientId>-clock-recovery- prefixed groups (READ, DESCRIBE, DELETE) | | transactions | <clientId>-tx transactional id | | autoCreateTopics | cluster CREATE (avoid in production) |

toMskIamPolicy() maps Kafka operations to kafka-cluster:* actions, turns prefixed patterns into name* ARN wildcards, and always includes kafka-cluster:Connect. Review both outputs against your organisation's least-privilege standards and current AWS documentation before applying — they are a starting point, not a rubber stamp.

Environment configuration

Build client and consumer configuration from environment variables with a strict precedence rule: explicit code options > env vars > built-in library defaults. The helpers only feed values in — anything you hard-code always wins, and any variable left unset keeps the library default.

The library never reads a .env file itself. Load one first with Node's built-in node --env-file=.env (Node 20.6+) or the dotenv package, then call the helpers:

import { KafkaClient, kafkaClientConfigFromEnv } from '@drarzter/kafka-client/core';

const { clientId, groupId, brokers, options } = kafkaClientConfigFromEnv();

const kafka = new KafkaClient(
  clientId ?? 'my-svc',           // env value or your fallback
  groupId ?? 'my-grp',
  brokers ?? ['localhost:9092'],
  {
    ...options,                   // only the keys whose env vars were present
    onMessageLost: alerting,      // code-level value — always applied, not env-configurable
  },
);

kafkaClientConfigFromEnv(env?, prefix?) reads KAFKA_-prefixed variables (CLIENT_ID, GROUP_ID, BROKERS, AUTO_CREATE_TOPICS, STRICT_SCHEMAS, NUM_PARTITIONS, TRANSACTIONAL_ID, CLOCK_RECOVERY_*, LAG_THROTTLE_*, and the security vars SSL, SASL_MECHANISM, SASL_USERNAME, SASL_PASSWORD, ALLOW_INSECURE). It returns { clientId?, groupId?, brokers?, options }, emitting only the keys whose variables were set. Malformed booleans/numbers/enums throw with the offending variable named. oauthbearer cannot come from env — token providers are functions, so configure them in code.

consumerOptionsFromEnv(env?, prefix?) reads KAFKA_CONSUMER_-prefixed variables into a Partial<ConsumerOptions> (retry, DLQ, deduplication, circuit breaker, TTL, GROUP_INSTANCE_ID, and more). Merge it under your code-level options with mergeConsumerOptions(), which applies the precedence rule — later layers win, and the nested objects (retry, deduplication, circuitBreaker, subscribeRetry) are deep-merged so a code layer can override a single field:

import { consumerOptionsFromEnv, mergeConsumerOptions } from '@drarzter/kafka-client/core';

const envDefaults = consumerOptionsFromEnv();
await kafka.startConsumer(
  ['orders'],
  handler,
  mergeConsumerOptions(envDefaults, { dlq: true }), // code layer wins on conflict
);

Both helpers accept an explicit env object (handy in tests) and a custom variable prefix. See docs/configuration.md for the full variable reference and .env.example for a ready-to-copy template.

Options reference

Send options

Options for sendMessage() — the third argument:

| Option | Default | Description | | ------ | ------- | ----------- | | key | — | Partition key for message routing | | headers | — | Custom metadata headers (merged with auto-generated envelope headers) | | correlationId | auto | Override the auto-propagated correlation ID (default: inherited from ALS context or new UUID) | | schemaVersion | 1 | Schema version for the payload | | eventId | auto | Override the auto-generated event ID (UUID v4) | | compression | — | Compression codec for the message set: 'gzip', 'snappy', 'lz4', 'zstd'; omit to send uncompressed | | deliverAfterMs | — | Delay delivery by at least this many milliseconds via a <topic>.delayed staging topic; requires a running startDelayedRelay() (see Delayed delivery) |

sendBatch() accepts compression and deliverAfterMs as top-level options (not per-message); all other options are per-message inside the array items.

Consumer options

| Option | Default | Description | | ------ | ------- | ----------- | | groupId | constructor value | Override consumer group for this subscription | | fromBeginning | false | Read from the beginning of the topic | | autoCommit | true | Auto-commit offsets | | retry.maxRetries | — | Number of retry attempts | | retry.backoffMs | 1000 | Base delay for exponential backoff in ms | | retry.maxBackoffMs | 30000 | Maximum delay cap for exponential backoff in ms | | dlq | false | Send to {topic}.dlq after all retries exhausted — message carries x-dlq-* metadata headers | | retryTopics | false | Route failed messages through per-level topics ({topic}.retry.1, {topic}.retry.2, …) instead of sleeping in-process; exactly-once routing semantics within the retry chain; requires retry (see Retry topic chain) | | interceptors | [] | Array of before/after/onError hooks | | retryTopicAssignmentTimeoutMs | 10000 | Timeout (ms) to wait for each retry level consumer to receive partition assignments after connecting; increase for slow brokers | | handlerTimeoutMs | — | Log a warning if the handler hasn't resolved within this window (ms) — does not cancel the handler | | deduplication.strategy | 'drop' | What to do with duplicate messages: 'drop' silently discards, 'dlq' forwards to {topic}.dlq (requires dlq: true — without it the client logs an error rather than discarding silently), 'topic' forwards to {topic}.duplicates | | deduplication.duplicatesTopic | {topic}.duplicates | Custom destination for strategy: 'topic' | | deduplication.store | in-memory | Pluggable DedupStore for the per-partition last-processed clock; supply a persistent store (e.g. Redis) so dedup survives restarts/rebalances (see Pluggable deduplication store) | | messageTtlMs | — | Drop (or DLQ) messages older than this many milliseconds at consumption time; evaluated against the x-timestamp header; see Message TTL | | circuitBreaker | — | Enable circuit breaker with {} for zero-config defaults; see Circuit breaker | | circuitBreaker.threshold | 5 | Failed handler attempts within windowSize that open the circuit | | circuitBreaker.recoveryMs | 30_000 | Milliseconds to wait in OPEN state before entering HALF_OPEN | | circuitBreaker.windowSize | threshold × 2, min 10 | Sliding window size in messages | | circuitBreaker.halfOpenSuccesses | 1 | Consecutive successes in HALF_OPEN required to close the circuit | | queueHighWaterMark | unbounded | Max messages buffered in the consume() iterator queue before the partition is paused; resumes at 50% drain. Only applies to consume() | | batch | false | (decorator only) Use startBatchConsumer instead of startConsumer | | partitionAssigner | 'cooperative-sticky' | Partition assignment strategy: 'cooperative-sticky' (minimal movement on rebalance, best for horizontal scaling), 'roundrobin' (even distribution), 'range' (contiguous partition ranges) | | groupInstanceId | — | Static group membership (group.instance.id) — a member that restarts within session.timeout.ms rejoins with the same partitions and no rebalance. Must be unique per member; not propagated to retry companions. See Static group membership | | transactionalIdPrefix | — | Override the base for this consumer's EOS transactional.ids. Prefer setting processId at the client level; use this only to override one specific consumer. See EOS fencing model | | onTtlExpired | — | Per-consumer override of the client-level onTtlExpired callback; takes precedence when set. Receives TtlExpiredContext — same shape as the client-level hook | | onMessageLost | — | Per-consumer override of the client-level onMessageLost callback; takes precedence when set. Use for consumer-specific dead-message alerting or structured logging | | onRetry | — | Per-consumer retry callback; fires in addition to the built-in metrics hook (does not replace it). Same signature as KafkaInstrumentation.onRetry | | subscribeRetry.retries | 5 | Max attempts for consumer.subscribe() when topic doesn't exist yet | | subscribeRetry.backoffMs | 5000 | Delay between subscribe retry attempts (ms) |

Module options

Passed to KafkaModule.register() or returned from registerAsync() factory:

| Option | Default | Description | | ------ | ------- | ----------- | | clientId | — | Kafka client identifier (required) | | groupId | — | Default consumer group ID (required) | | brokers | — | Array of broker addresses (required) | | name | — | Named client identifier for multi-client setups | | isGlobal | false | Make the client available in all modules without re-importing | | autoCreateTopics | false | Auto-create topics on first send (dev only) | | numPartitions | 1 | Number of partitions for auto-created topics | | strictSchemas | true | Validate string topic keys against schemas registered via TopicDescriptor | | security | — | TLS + SASL transport security with secure-by-default rules ({ ssl, sasl, allowInsecure }); see Transport security | | instrumentation | [] | Client-wide instrumentation hooks (e.g. OTel). Applied to both send and consume paths | | transactionalId | ${clientId}-tx, or ${clientId}.${processId}-tx when processId is set | Transactional producer ID for transaction() calls. Must be unique per replica — transaction() commits no offsets, so it is not covered by consumer-group fencing and this ID is the only thing separating two instances. The client logs a warning when the same ID is registered twice within one process | | processId | generated UUID | Stable identity of this process, used as the base for every EOS transactional.id the consumers create. Set it to something unique per replica (a StatefulSet pod name, a UUID you persist). See EOS fencing model | | onMessageLost | — | Called when a message is silently dropped without DLQ — use to alert, log to external systems, or trigger fallback logic | | onTtlExpired | — | Called when a message is dropped due to TTL expiration (messageTtlMs) and dlq is not enabled; receives { topic, ageMs, messageTtlMs, headers } | | onRebalance | — | Called on every partition assign/revoke event across all consumers created by this client | | clockRecovery.topics | — | Topics to scan on connectProducer() to recover the highest x-lamport-clock, so the clock stays monotonic across restarts (see Deduplication) | | clockRecovery.timeoutMs | 30000 | Max time (ms) to wait for clock recovery before proceeding with a partial result | | lagThrottle | — | Delay sends when a consumer group's lag exceeds maxLag (see Lag-based producer throttling) | | lagThrottle.maxLag | — | Lag threshold (messages) above which sends are delayed (required when lagThrottle is set) | | lagThrottle.groupId | default group | Consumer group whose lag is monitored | | lagThrottle.pollIntervalMs | 5000 | How often (ms) to poll getConsumerLag() in the background | | lagThrottle.maxWaitMs | 30000 | Max time (ms) a send waits while throttled before proceeding anyway (best-effort, not hard back-pressure) | | transport | ConfluentTransport | Custom KafkaTransport implementation — target an alternative broker library or inject a deterministic fake in tests |

Advanced — direct transport access. ConfluentTransport and the full KafkaTransport interface family (IProducer, IConsumer, IAdmin, …) are exported from @drarzter/kafka-client/core. When you need low-level admin operations the facade does not expose (e.g. per-partition watermarks), build a transport instead of deep-importing the raw driver:

import { ConfluentTransport } from '@drarzter/kafka-client/core';

const admin = new ConfluentTransport('ops-cli', brokers).admin();
await admin.connect();
const watermarks = await admin.fetchTopicOffsets('orders'); // [{ partition, low, high }]

Module-scoped (default) — import KafkaModule i