kafka-harbor

The application layer that Kafka clients don't give you: retry topics, dead-letter queue, graceful shutdown, serialization and observability for Node.js. Client-agnostic by design.

Messages cross the sea; the harbor is where they dock safely. Every Kafka client for Node.js stops at the protocol: you get a producer, a consumer, and good luck. Retry with backoff, a dead-letter queue, offsets committed only after your code ran, a shutdown that does not lose or duplicate work: all of it gets rebuilt by hand in every project. kafka-harbor is that layer, done once, on top of the client you already use.

Status: 0.x. The API below is the intended 1.0 surface and is exercised end to end against a real broker. Until 1.0.0, a minor release may still adjust it; every change is listed in CHANGELOG.md.

import { createHarbor } from 'kafka-harbor'
import { confluentAdapter } from 'kafka-harbor/adapters/confluent'

const harbor = createHarbor({
  clientId: 'orders-service',
  brokers: ['kafka-1:9092', 'kafka-2:9092'],
  adapter: confluentAdapter()
})

const consumer = harbor.consumer({
  groupId: 'orders-workers',
  retry: { levels: [{ delay: '5s' }, { delay: '1m' }, { delay: '10m' }] },
  autoCreateTopics: true
})

consumer.subscribe<Order>('orders', async (message, ctx) => {
  await fulfill(message.value)      // throws? -> orders-retry-1, then -retry-2, -retry-3, then orders-dlq
  ctx.logger.info(`order ${message.key} done on attempt ${ctx.attempt}`)
})

await consumer.start()
harbor.enableSignalHandlers()        // SIGTERM -> finish in-flight handlers, commit, leave, disconnect

Table of contents

• Why another Kafka library?
• What it does not do
• Install
• Core concepts
• Durations
• Producer
• Consumer
• Retry topics and the DLQ
• Graceful shutdown
• Health
• Idempotency
• Observability
• Serialization
• Schema Registry
• Headers
• Events
• Errors
• Adapters
• NestJS and decorators
• Testing your handlers
• Development

Why another Kafka library?

It is not a client. kafka-harbor runs on top of a client through a small ClientAdapter interface (connect, produce, consume, commit, pause, resume, admin). Two adapters ship with it: the default wraps @confluentinc/kafka-javascript, Confluent's supported client with librdkafka underneath, and kafka-harbor/adapters/platformatic wraps @platformatic/kafka, a pure TypeScript client with no native binding. The interface is designed so that the core never sees a client type, and the same contract suite runs against both.

Every Node.js team using Kafka ends up writing the same application layer on top of whichever client they picked, because the clients stop at the protocol. The comparison below is against the clients themselves, which is the honest one: kafka-harbor is not a replacement for them, it runs on top of one.

| | kafka-harbor | @confluentinc/kafka-javascript | kafkajs | @platformatic/kafka | |---|---|---|---|---| | Retry topics with a delay per level, tracking headers, retryIf predicate | ✅ orders-retry-1..N, delays honored, headers validated as network input | ➖ you build it | ➖ you build it | ➖ you build it | | Dead-letter topic with the original bytes and the failure trail | ✅ automatic after the last level | ➖ | ➖ | ➖ | | Draining the DLQ back into service | ✅ harbor.redrive(), resumable, filterable | ➖ | ➖ | ➖ | | Offset committed only after the retry/DLQ produce was acknowledged | ✅ by construction; a failed produce stops the consumer instead of committing | ➖ your ordering | ➖ your ordering | ➖ your ordering | | Graceful shutdown: wait for handlers with a deadline, commit, leave, disconnect, report what was abandoned | ✅ ShutdownTimeoutError names the count | ➖ disconnect() waits for the running handler, no deadline, no report | ➖ same | ➖ close() | | harbor.abort(): stop without committing when reprocessing is the right call | ✅ | ➖ throw and hope auto-commit is off | ➖ | ➖ | | Serialization that never silently flattens (Map, undefined, NaN rejected) | ✅ strict JSON default, pluggable per topic, Schema Registry through kafka-harbor/schema-registry | ➖ bytes | ➖ bytes | ✅ pluggable serdes, schema registry | | In-memory broker for unit tests, same core code, no Docker | ✅ kafka-harbor/testing | ❌ | ❌ | ❌ | | Client-agnostic: swap the client without touching handlers | ✅ ClientAdapter, contract suite for authors | n/a | n/a | n/a | | Typed outcome events (messageRetried, messageDeadLettered, ...) with correlation id | ✅ | ➖ client events | ➖ instrumentation events | ➖ | | Health snapshot for probes | ✅ isHealthy() / health() | ➖ | ➖ | ➖ | | Idempotent producer and acks=all on by default | ✅ set by the adapter, cannot be overridden by accident | ➖ opt-in | ➖ opt-in | ➖ opt-in | | Runtime dependencies | breakwater + the client you choose | native librdkafka | none (pure JS) | none (pure TS) |

The rows are not a knock on the clients: transactions, exactly-once, the schema registry client itself, fetch tuning and wire performance are theirs, and the Confluent client is the default underneath and the platformatic client the pure TypeScript alternative, both behind the same adapter contract. The rows are the layer every project rebuilds by hand, done once, with the ordering guarantees tested against a real broker.

The design principle behind every decision: losing a message is never the default. Every failure ends in a retry topic, in the DLQ, or in an explicit stop of the consumer. There is no silent path.

What it does not do

• At-least-once only. Duplicates are possible after a crash between handler and commit, a rebalance mid-handler, or an abandoned shutdown; docs/delivery-semantics.md lists every case. Exactly-once effects come from idempotency: an engine that runs the handler once per key.
• A retry delay must fit under the poll interval (maxProcessingTime, default 5 minutes), because the retry consumer waits the delay before the handler runs. The Confluent adapter sets the client's max.poll.interval.ms from it; a longer ladder needs a longer maxProcessingTime.
• Retry breaks ordering. A message that goes through a retry topic is processed after later messages on the original topic. The alternative, blocking the partition until it succeeds, is what harbor.abort() gives you.
• Durations top out at about 24.8 days (2147483647 ms), the longest a timer can hold. A longer shutdown timeout or retry delay is a ConfigError, not a wait that ends after a millisecond.
• The default adapter has a native dependency. @confluentinc/kafka-javascript ships prebuilt binaries for Node 18 to 24 on Linux (glibc and musl, x64 and arm64), macOS and Windows; on Node 26 it compiles librdkafka at install and needs a build toolchain in the image. Docker images lists what was verified. kafka-harbor/adapters/platformatic needs no binary at all, and any other client can be plugged in through ClientAdapter.
• No transactions. The producer is idempotent and the consumer commits after the handler; there is no sendOffsetsToTransaction, so a handler that produces and consumes is at-least-once on both sides.

Install

npm install kafka-harbor @confluentinc/kafka-javascript

or, with the pure TypeScript client:

npm install kafka-harbor @platformatic/kafka

The clients are optional peer dependencies: install the one behind the adapter you use (kafka-harbor/adapters/confluent or kafka-harbor/adapters/platformatic). Node.js >= 22 is required by kafka-harbor itself.

Docker images

@confluentinc/kafka-javascript 1.10 downloads a prebuilt binary at install time when one exists for the platform, and compiles librdkafka from source otherwise. Its release publishes binaries for Node 18, 20, 21, 22, 23 and 24 (ABI 108 to 137) on Linux glibc and musl (x64 and arm64), macOS (x64 and arm64) and Windows (x64). There is no binary for Node 26 (ABI 147).

What was verified with npm install @confluentinc/[email protected] followed by loading the module, on 2026-09-06:

| Image | linux/arm64 | linux/amd64 | Outcome | |---|---|---|---| | node:22-bookworm-slim | yes | yes | prebuilt binary, no toolchain needed | | node:22-alpine | yes | | prebuilt binary (musl), no toolchain needed | | node:24-bookworm-slim | yes | | prebuilt binary, no toolchain needed | | node:24-alpine | yes | yes | prebuilt binary (musl), no toolchain needed | | node:26-bookworm-slim | yes | | install fails: no binary, and the image has no compiler | | node:26-bookworm | yes | | install fails: the build downloads zlib, OpenSSL, zstd and libcurl sources and the image has no curl or wget | | node:26-bookworm-slim + librdkafka-dev from Confluent's apt repository, CKJS_LINKING=dynamic BUILD_LIBRDKAFKA=0 | yes | | works, install in 78s: the binding links against the system librdkafka 2.15.0 instead of compiling one | | node:26-alpine + librdkafka-dev from Alpine's own repository, CKJS_LINKING=dynamic BUILD_LIBRDKAFKA=0 | yes | | works, install in about 2 minutes: links against Alpine's librdkafka 2.14.1 | | node:26-bookworm-slim and node:26-alpine with a compiler, python3, make, curl, perl and patch, default static build | yes | | install fails after about 10 minutes of compiling: ar: /probe/node_modules/: file format not recognized while merging librdkafka-static.a |

Confluent tracks Node 26 prebuilt binaries in confluentinc/confluent-kafka-javascript#397. Until they ship, the route that works on Node 26 is the one the client documents for unsupported platforms: link the binding dynamically against a librdkafka installed from a package repository, instead of letting the install compile librdkafka and its dependencies statically. Debian:

FROM node:26-bookworm-slim
RUN apt-get update && apt-get install -y --no-install-recommends curl gnupg ca-certificates python3 make g++ pkg-config \
 && install -d /etc/apt/keyrings \
 && curl -fsSL https://packages.confluent.io/deb/8.0/archive.key | gpg --dearmor -o /etc/apt/keyrings/confluent.gpg \
 && echo "deb [signed-by=/etc/apt/keyrings/confluent.gpg] https://packages.confluent.io/clients/deb bookworm main" > /etc/apt/sources.list.d/confluent-clients.list \
 && apt-get update && apt-get install -y --no-install-recommends librdkafka-dev \
 && rm -rf /var/lib/apt/lists/*
ENV CKJS_LINKING=dynamic BUILD_LIBRDKAFKA=0
COPY package*.json ./
RUN npm ci

Alpine, whose own repository carries a recent librdkafka:

FROM node:26-alpine
RUN apk add --no-cache python3 make g++ pkgconfig librdkafka-dev
ENV CKJS_LINKING=dynamic BUILD_LIBRDKAFKA=0
COPY package*.json ./
RUN npm ci

The runtime image then needs the shared library the binding was linked against (librdkafka1 from the Confluent repository on Debian, librdkafka on Alpine), not the compiler; a multi-stage build copies node_modules from the build stage. librdkafka-dev from Debian's own repository is older than what the client bundles; the Confluent repository carries the matching one. On Node 22 and 24 none of this is needed: the prebuilt binary is downloaded and the official images work as they are.

The default static build, the one npm install attempts on its own when no binary exists, cannot complete on Linux from an npm install: librdkafka merges its static dependencies with a GNU ar MRI script (ADDLIB /path/to/lib.a), and GNU ar (binutils 2.40 verified) cuts such a path at an @, which every scoped package path (node_modules/@confluentinc/...) contains. The same script with the @ removed from the path succeeds. macOS builds from source because Apple's libtool is used there instead of ar, which is why a Node 26 install works on a Mac with Xcode's command line tools and fails in a Linux container with the same toolchain. None of this applies to kafka-harbor/adapters/platformatic: @platformatic/kafka is TypeScript all the way down and installs on every image Node runs on.

Core concepts

• Harbor: the entry point. Holds the connection, the serializer, the header names and the logger. createHarbor() does not connect; the first send() or start() does, or call connect().
• Producer: harbor.producer(). Serializes, adds the automatic headers, produces with acks=all and an idempotent producer, and retries transient broker failures through breakwater.
• Consumer: harbor.consumer({ groupId }). One consumer group, one or more topics, one handler per topic. Owns the retry ladder and the DLQ for each topic it subscribes to.
• Adapter: the client behind it all. Explicit in the config so that the core has no dependency on any client.
• Message: what the handler receives. Deserialized value, string key, string headers, Date timestamp, and retry metadata when it came through a retry topic.

Durations

Every option that is a span of time (shutdown(), enableSignalHandlers(), maxProcessingTime, the delay of a retry level, the redrive idleTimeout) takes a number of milliseconds or a string with a unit:

| Suffix | Unit | Example | |---|---|---| | ms | milliseconds | '250ms' | | s | seconds | '30s' | | m | minutes | '1m', '1.5m' | | h | hours | '2h' | | d | days | '1d' |

Decimals are accepted, a space before the unit is tolerated, and a bare number in a string ('30') is refused: it is more likely a forgotten unit than thirty milliseconds. The ceiling is 2147483647 ms, about 24.8 days, the longest a timer can hold; anything above is a ConfigError at the call, never a wait that ends a millisecond later. parseDuration() and MAX_DURATION_MS are exported for code that wants the same rule.

Producer

const producer = harbor.producer<Order>()

await producer.send('orders', { key: order.id, value: order })

await producer.sendBatch('orders', [
  { key: 'a', value: orderA, headers: { 'x-tenant': 'acme' } },
  { key: 'b', value: orderB, partition: 3 }        // explicit partition, rarely needed
])

// A serializer for this producer only, and a tighter retry ladder.
import { exponential } from 'breakwater'

const events = harbor.producer<OrderEvent>({
  serializer: avroSerializer(schema),
  retry: { attempts: 3, backoff: exponential({ initial: 50, max: 1_000 }) }
})

• Every message gets x-correlation-id (kept if you set a non-blank one; minted otherwise, also when the library forwards a message to a retry topic, the DLQ or back from it), x-produced-at and x-producer (your clientId).
• value: null is a tombstone: the record goes out with no value at all, which a compacted topic reads as "delete this key", and the serializer is never asked. Handlers receive tombstones as value: null.
• Keyed messages land on the partition Kafka's default partitioner picks (murmur2). partitionForKey(key, partitions) computes the same number, for code that needs to know where a key goes: sharding a cache by partition, asserting co-location of related keys, or routing an unkeyed message next to a keyed one.
• A batch is serialized before any byte leaves the process: one unencodable value means nothing is produced.
• send() resolves after the broker acknowledged. Transient failures are retried (default: 5 attempts, exponential backoff with full jitter); a failure marked retryable: false is not. When the attempts run out you get breakwater's RETRY_EXHAUSTED with the last failure as cause.

Consumer

const consumer = harbor.consumer({
  groupId: 'orders-workers',

  retry: {
    levels: [{ delay: '5s' }, { delay: '1m' }, { delay: '10m' }],
    retryIf: (error) => !(error instanceof ValidationError),  // default: everything unless retryable === false
    topicNaming: (topic, level) => `${topic}-retry-${level}`   // default
  },
  dlq: {
    enabled: true,                                              // default
    topicNaming: (topic) => `${topic}-dlq`                      // default
  },

  concurrency: 4,             // partitions processed at once; order is kept within each. Default: 1
  autoCreateTopics: true,     // create the retry and DLQ topics through the Admin API on start. Default: false
  topicDefaults: { partitions: 3, replicationFactor: 3 },
  fromBeginning: false,       // where a brand-new group starts. Default: false (latest)
  maxProcessingTime: '5m'     // every retry delay must fit under it; see below. Default: '5m'
})

consumer
  .subscribe<Order>('orders', onOrder)
  .subscribe<Payment>('payments', onPayment, { serializer: protobufSerializer(Payment) })

await consumer.start()

The handler signature is (message, context):

consumer.subscribe<Order>('orders', async (message, ctx) => {
  message.topic       // 'orders' or 'orders-retry-2'
  message.partition   // number
  message.offset      // string
  message.key         // string | null
  message.value       // Order, deserialized
  message.headers     // Record<string, string>
  message.timestamp   // Date
  message.retry       // { count, originalTopic, firstFailureAt, lastError } on a retry topic

  ctx.groupId         // the consumer group the handler runs in
  ctx.correlationId   // from the headers, if any
  ctx.attempt         // 1 on first delivery, retry count + 1 afterwards
  ctx.logger          // the harbor's logger
  ctx.signal          // aborts when shutdown gave up waiting for this handler
})

What happens next depends on how the handler ends, and nothing else. There is no ack callback to forget:

| The handler... | The consumer... | |---|---| | returns | commits the offset | | throws, and retryIf(error) is true, and a retry level is left | produces to the next retry topic, then commits | | throws otherwise | produces to the DLQ, then commits | | throws harbor.abort(error) | stops without committing (infrastructure bug: reprocess after restart); handlers running on other partitions finish and commit first | | throws, with no retry level left and the DLQ disabled | stops without committing and emits error |

The retry or DLQ produce is acknowledged by the broker before the source offset is committed. If it is not acknowledged, the consumer stops and the message stays where it is: it will be redelivered. A commit that fails after the work is safe (a rebalance in progress, for instance) is reported through the error event and the consumer carries on; that message is redelivered too.

Batches

consumer.subscribeBatch<Order>('orders', async (messages, ctx) => {
  await bulkInsert(messages.map((message) => message.value))   // ctx: { groupId, topic, partition, logger, signal }
}, { size: 100, maxWait: '1s' })

A batch is up to size consecutive messages of one partition, in offset order, or fewer once maxWait passed since the first one arrived. Resolving commits the offset after the last message, once. Throwing sends every message of the batch down the retry ladder with that error, each with its own tracking headers; throwing new BatchFailedError(failedMessages, cause) sends only the messages it names and commits the rest as processed. On a retry topic the messages wait their delay and run as a batch again. A message that does not deserialize is dead-lettered on its own and the rest of the batch runs.

One batch runs per partition at a time and at most size messages wait behind it, so offsets never commit out of order and a delivery never outlives one batch run (keep a batch under maxProcessingTime). A batch still collecting when the consumer stops is left uncommitted for the next member; one collecting when the partition is taken away runs first, like a handler already running. Events are per message plus one batchProcessed per run; a message from a batch carries batch (the size) and the batch's durationMs, and the metrics entry points measure batch duration in its own histogram. A batch subscription cannot share a retry topic with another subscription. Idempotency does not apply to batches: deduplicate inside the handler, or use another consumer.

Retry topics and the DLQ

A failed message is re-produced, bytes untouched, to orders-retry-1. The same consumer group also consumes orders-retry-1, waits until the message is 5s old, and runs the handler again. Fail again: orders-retry-2, 1m. And so on until the ladder is exhausted, then orders-dlq.

Each hop rewrites the tracking headers:

| Header | Meaning | |---|---| | x-retry-count | handler failures so far | | x-original-topic | where the message was first produced | | x-first-failure-at | ISO-8601 instant of the first failure | | x-last-error | description of the latest failure, bounded to 1 KiB | | x-dead-lettered-at | set on the DLQ hop only | | x-redriven-from, x-redriven-at | set by harbor.redrive() when a message comes back from the DLQ |

These headers come from the network and are validated before use: a blank or corrupt count never turns into 0 or NaN; the whole block is discarded and the message counts as a first delivery on that level. A message that fails to deserialize goes straight to the DLQ; retrying would not decode it either.

Things to know:

• Delays are bounded by maxProcessingTime (default 5 minutes). A retry consumer waits the delay before the handler runs; a wait longer than the client's poll interval would get it kicked out of the group, so the adapter receives the same number (the Confluent adapter sets max.poll.interval.ms from it unless your passthrough pins another value). A level above the bound is a ConfigError at construction naming retry.levels[i].delay.
• Each retry level is a group member of its own. A consumer with two levels joins its group three times: once for the original topics, once per level. A message sleeping out its delay on orders-retry-2 never holds a worker that orders or orders-retry-1 is waiting for, whatever concurrency is. Kafka assigns each topic among the members subscribed to it, so the members of one group may consume different topics.
• Retention must exceed the delay. A message with a 1h delay on a topic with 30 minutes of retention is a lost message. topicDefaults and your own topic configs are yours to set accordingly.
• Naming uses hyphens (orders-retry-1, orders-dlq), the same as Spring Kafka's defaults, because Kafka warns that . and _ collide in metric names. Both naming functions are configurable.
• One ladder per topic by default; share it by naming. Three levels times twenty topics is sixty retry topics. A naming function that returns one name per level makes that level's topic shared by every subscription of the consumer, and a DLQ naming function that returns one name shares the DLQ: retry: { topicNaming: (_topic, level) => \orders-service-retry-${level}` }, dlq: { topicNaming: () => 'orders-service-dlq' }turns sixty topics into four. On a shared topic thex-original-topicheader decides which handler a message belongs to, so the delay of a level stays the same for every topic on it. A message there whose header names a topic the consumer does not subscribe to goes to the shared DLQ (the consumer stops instead when the owners have different DLQs). A shared name may not cross levels or name a topic the consumer consumes; both areConfigErroratsubscribe()`.
• With no levels configured (the default), a failure goes straight to the DLQ.

Draining the DLQ back into service

const result = await harbor.redrive({
  from: 'orders-dlq',
  to: 'orders',               // default: each message's own x-original-topic header
  groupId: 'orders-dlq-redrive', // default: `${from}-redrive`; the offset persists between runs
  max: 500,                    // stop after this many; default: no limit
  idleTimeout: '5s',           // stop once nothing arrived for this long; default
  filter: (message) => message.retry?.lastError !== 'ValidationError: bad sku'  // false skips (committed, not re-injected)
})
result // { from: 'orders-dlq', reprocessed: 498, skipped: 2 }

Each message is re-produced with its original key and value; the failed run's tracking headers are removed so it starts a fresh retry ladder, and x-redriven-from / x-redriven-at record the operation. The DLQ offset is committed only after the broker acknowledged the re-produce, so an interrupted redrive resumes where it stopped. A messageRedriven event fires per message. A message without x-original-topic fails the run unless to is given; a filter that throws, a re-produce that is not acknowledged, or a serializer reporting a transient failure (retryable: true, a schema registry that is away) stops the run with that error and leaves the message uncommitted, so running it again picks up there.

Idempotency

At-least-once delivery means a handler can see the same message twice (the list of cases is short and honest). idempotency runs every handler of a consumer through an engine that executes once per key and replays the first outcome for a repeat, which is then committed without the handler running again. quayside fits the seam as it is, with its storages (memory, Redis, Postgres, MySQL, DynamoDB), fencing and replay window:

import { createHarbor } from 'kafka-harbor'
import { confluentAdapter } from 'kafka-harbor/adapters/confluent'
import { Idempotency } from 'quayside'
import { RedisStorage } from 'quayside/redis'

const harbor = createHarbor({ clientId: 'orders-service', brokers, adapter: confluentAdapter() })
const engine = new Idempotency({
  storage: new RedisStorage(redis),
  namespace: 'orders-service',
  onConflict: 'wait',      // a redelivery racing a live run waits for it and replays
  lockTtl: '5m'            // at least the longest a handler may run (maxProcessingTime)
})

// Every handler of the consumer, keyed by the delivery: `${groupId}:${topic}:${partition}:${offset}`.
const consumer = harbor.consumer({ groupId: 'orders-service', idempotency: { engine } })
// One topic keyed by the business id, payload fingerprinted: the same id with other content is refused.
consumer.subscribe<Order>('orders', handleOrder, {
  idempotency: { engine, key: (message) => ({ key: `order:${message.value.id}`, payload: message.value }) }
})

• Key. The default names the delivery (group, topic, partition, offset) and collapses exactly the redeliveries at-least-once allows; it is exported as defaultIdempotencyKey for a key that builds on it. A business key collapses duplicates the producer sent as well. Returning { key, payload } has the engine fingerprint the payload and refuse the same key with different content; { key, resultTtl } sets the replay window for that message.
• Outcomes. A replay is reported as messageProcessed with replayed: true and counted in messages_replayed_total. A handler failure is not stored: the retry runs the handler again. An error from the engine itself (storage unavailable, key refused, key function threw) is a handler failure and walks the ladder like any other; quayside's errors carry a code, so retry.retryIf can send a deterministic refusal such as IDEMPOTENCY_KEY_REUSE straight to the DLQ.
• Conflicts and leases. A redelivery that arrives while the first run is still executing is a conflict: with quayside's onConflict: 'wait' it waits for the winner and replays; with 'reject' it fails and walks the ladder. The engine's lock must outlive the handler (lockTtl above maxProcessingTime), or a run that finishes after its lease expired is refused and retried.
• persistFailures. quayside stores every failure under the key, transient ones included. With the per-delivery key that is what you want: a redelivery of a failed delivery replays the failure with its retryable flag, so a business rejection reaches the DLQ without the work running twice. With a business key the retry hops share the key and would replay the failure instead of retrying, so keep failures unpersisted there.
• Any engine. The seam is one method, executeWithMetadata(input, run), resolving { value, replayed }; IdempotencyEngine is exported for an engine of your own. The core imports nothing.

Graceful shutdown

await harbor.shutdown('30s')   // or harbor.enableSignalHandlers() for SIGTERM/SIGINT

1. Every consumer stops taking new messages. Messages waiting for a retry delay are released at once, uncommitted.
2. Handlers already running get up to the timeout to finish. The ones that finish commit their offsets on the way out.
3. Handlers still running when the timeout elapses are abandoned: their ctx.signal aborts, their offsets are not committed (the messages will be redelivered), and shutdown() rejects with ShutdownTimeoutError after everything else is done. At-least-once, said out loud.
4. Consumers leave their groups, then the client disconnects.

Health

harbor.isHealthy()   // boolean, for a liveness probe
harbor.health()      // { healthy, state, adapter, consumers: [{ groupId, status, stoppedBecause }] }

Synchronous and cheap: it reads the state the harbor already tracks and never calls the broker. A harbor is healthy until it shuts down or until a consumer stops on its own (stoppedBecause is 'abort' or 'crash'); a consumer stopped by shutdown() does not count against it. Connection is lazy by design, so a harbor that has not connected yet is healthy.

Observability

Three things feed every dashboard: the typed events, consumer lag, and two entry points that turn both into Prometheus metrics or OpenTelemetry signals. The core imports neither client; prom-client and @opentelemetry/api are optional peer dependencies, installed only by the application that uses the matching entry point.

Lag

await consumer.lag()   // [{ groupId, topic, partition, low, high, committed, lag }], every consumed topic, retry ladder included
await harbor.lag()     // the same for every running consumer

Lag is the number of records between the group's position and the high watermark of the partition. The position is the committed offset while there is one inside the partition's range. Until the first commit, or when the committed offset fell out of range (the records expired, the topic was recreated), it is where the group would start: the first offset still held with fromBeginning, the high watermark otherwise. One round of admin calls per call, never per message: a metrics scrape is the caller this is meant for, and concurrent harbor.lag() calls share one round. The adapter must report offsets (admin.fetchTopicOffsets and admin.fetchCommittedOffsets; the Confluent and memory adapters do); one that does not makes lag() reject with a ConfigError naming the capability. A partition without a leader yet (right after the topic was created) is left out until it has one. harbor.lag() covers the running consumers: one whose offsets could not be fetched is reported through the error event with the adapter scope and left out, and the call rejects only when no consumer answered.

Prometheus

import { Registry } from 'prom-client'
import { prometheusMetrics } from 'kafka-harbor/prometheus'

const registry = new Registry()
const metrics = prometheusMetrics(harbor, { registry })   // serve registry.metrics() on your scrape endpoint
// later, if the harbor goes away before the process does:
metrics.detach()

| Metric | Labels | What it counts | |---|---|---| | kafka_harbor_messages_processed_total | group, topic | handler succeeded, offset committed | | kafka_harbor_messages_replayed_total | group, topic | of those, replayed by the idempotency engine without running the handler | | kafka_harbor_message_processing_duration_seconds | group, topic, outcome | handler duration per message, histogram; outcome is processed, retry, dead-letter, abort or crash; messages handled in a batch are measured by the batch histogram instead | | kafka_harbor_batches_processed_total | group, topic, outcome | subscribeBatch handler runs | | kafka_harbor_batch_processing_duration_seconds | group, topic, outcome | batch handler duration, histogram | | kafka_harbor_messages_failed_total | group, topic, outcome | handler failures by what happened next | | kafka_harbor_messages_retried_total | group, topic, level | messages forwarded to a retry topic | | kafka_harbor_messages_dead_lettered_total | group, topic | messages forwarded to the DLQ, after the broker acknowledged | | kafka_harbor_messages_redriven_total | from, to | dead letters re-injected by redrive() | | kafka_harbor_messages_produced_total | topic, kind | records acknowledged by the broker; kind is send (harbor.producer()), retry, dead-letter or redrive | | kafka_harbor_produce_duration_seconds | topic, kind | produce call to acknowledgment, histogram | | kafka_harbor_errors_total | scope | error events: consumer, producer, adapter | | kafka_harbor_consumer_stops_total | group, reason | shutdown, abort or crash | | kafka_harbor_consumer_lag | group, topic, partition | gauge collected on scrape through harbor.lag() |

Options: registry (default: prom-client's global one), prefix (default kafka_harbor_, '' for none), buckets for the histograms in seconds (default 5ms to 10s), and lag: false to skip the gauge, which is required for an adapter that does not report offsets (otherwise a ConfigError at construction). A lag collection that fails leaves the gauge without series for that scrape and is reported through the harbor's error event; the scrape itself succeeds. detach() unsubscribes and removes the lag gauge; the counters and histograms stay registered until registry.clear(). Labels are deliberately low-cardinality: never an offset or a correlation id.

A Grafana dashboard built on these metrics is in examples/grafana/kafka-harbor.json: throughput, outcomes, dead letters, handler latency percentiles and lag per group and topic.

OpenTelemetry

import { createHarbor } from 'kafka-harbor'
import { otelMetrics, otelTracing } from 'kafka-harbor/otel'

const harbor = createHarbor({ clientId: 'orders-service', brokers, adapter: myAdapter(), instrumentation: otelTracing() })
const metrics = otelMetrics(harbor)   // instruments under kafka_harbor.*, the same signals as the Prometheus entry point

otelMetrics records the same signals as instruments named kafka_harbor.messages.processed, kafka_harbor.messages.replayed, kafka_harbor.message.processing.duration, kafka_harbor.batches.processed, kafka_harbor.batch.processing.duration, kafka_harbor.messages.failed, kafka_harbor.messages.retried, kafka_harbor.messages.dead_lettered, kafka_harbor.messages.redriven, kafka_harbor.messages.produced, kafka_harbor.produce.duration, kafka_harbor.errors, kafka_harbor.consumer.stops and the observable gauge kafka_harbor.consumer.lag, with kafka_harbor.group, kafka_harbor.topic, kafka_harbor.outcome, kafka_harbor.level, kafka_harbor.kind, kafka_harbor.scope, kafka_harbor.reason, kafka_harbor.from, kafka_harbor.to and kafka_harbor.partition attributes. Options: meterProvider, boundaries for the histograms in seconds (default 5ms to 10s) and lag: false, required for an adapter that does not report offsets. Start your SDK, or pass meterProvider, before calling it: the metrics API has no late-binding proxy. A lag collection that fails is reported through the harbor's error event.

otelTracing returns the instrumentation hooks the harbor calls around produce calls and handlers. Every produce call runs inside a PRODUCER span named <topic> send with a kafka_harbor.kind attribute (send, retry, dead-letter or redrive), and the span's context is written into each record's headers by the configured propagator (W3C traceparent and tracestate with the SDK's default) unless the record already carries one. Every handler runs inside a CONSUMER span named <topic> process, parented to the context read from the message headers (a batch handler gets one span, linked to every message's context, with messaging.batch.message_count), with the OpenTelemetry messaging attributes (messaging.system, messaging.destination.name, messaging.consumer.group.name, messaging.destination.partition.id, messaging.kafka.offset) plus kafka_harbor.attempt, kafka_harbor.correlation_id and kafka_harbor.original_topic. The handler runs under the extracted context, so baggage the producer propagated is active there too. A retry, DLQ or redrive hop is a PRODUCER span parented to the context the forwarded message carries, and the hop copies that message's headers, so the first attempt, every retry, the dead-lettering and the redrive of one message belong to the trace that produced it. A handler that throws marks its span with the exception and an error status.

A hook that throws, or an SDK that misbehaves, never changes what the pipeline does: the core logs the failure and runs the work unwrapped. Without an SDK registered the hooks are inert.

Your own instrumentation

instrumentation accepts any object with some of wrapProduce(batch, run), onProduce(record), wrapHandler(message, context, run) and wrapBatchHandler(messages, context, run); otelTracing() is one implementation. wrapProduce sees every produce call, the hops included: batch.kind says what it is for and, for a hop, batch.origin is the consumed message being forwarded, headers included. onProduce runs inside wrapProduce, so a context the wrapper sets up is what it sees, and whatever it returns is added to the record's headers, a header the record already carries taking precedence.

Serialization

The default is JSON, strict. JSON.stringify turns Map, Set, typed arrays, RegExp, Error and Promise into {} without a word, drops undefined inside objects, and encodes NaN as null. kafka-harbor rejects every one of those shapes with SerializationError before a byte is produced, so the handler always gets what the producer meant. Date is the single conversion accepted (encoded as ISO-8601, decoded as a string).

import { createHarbor, jsonSerializer, rawSerializer, stringSerializer, type Serializer } from 'kafka-harbor'
import { confluentAdapter } from 'kafka-harbor/adapters/confluent'

const harbor = createHarbor({ clientId: 'orders-service', brokers, adapter: confluentAdapter(), serializer: jsonSerializer() }) // default
harbor.producer({ serializer: rawSerializer() })                          // Buffer in, Buffer out
consumer.subscribe('logs', handler, { serializer: stringSerializer() })   // UTF-8 text

const msgpack: Serializer<MyType> = {
  serialize: (value, topic) => encode(value),
  deserialize: (bytes, topic) => decode(bytes)
}

A serializer applies harbor-wide, per producer, per consumer or per topic, most specific wins. Whatever deserialize throws is taken as malformed bytes and sent straight to the DLQ, unless the error carries retryable: true, the way a serializer backed by a service reports the service being away; that message walks the retry ladder instead. Either method may return a promise: the producer serializes every value of a batch before any byte leaves, and the consumer awaits the value before the handler runs. A serializer backed by a schema registry is the case in point: Schema Registry.

Schema Registry

import { AvroDeserializer, AvroSerializer, SchemaRegistryClient, SerdeType } from '@confluentinc/schemaregistry'
import { createHarbor } from 'kafka-harbor'
import { confluentAdapter } from 'kafka-harbor/adapters/confluent'
import { schemaRegistrySerializer } from 'kafka-harbor/schema-registry'

const registry = new SchemaRegistryClient({ baseURLs: ['http://schema-registry:8081'] })
const orders = schemaRegistrySerializer({
  serializer: new AvroSerializer(registry, SerdeType.VALUE, { useLatestVersion: true }),
  deserializer: new AvroDeserializer(registry, SerdeType.VALUE, {})
})
const harbor = createHarbor({ clientId: 'orders-service', brokers, adapter: confluentAdapter(), serializer: orders })

kafka-harbor/schema-registry fits the serdes of @confluentinc/schemaregistry to the harbor's Serializer (npm install @confluentinc/schemaregistry; it is an optional peer dependency). The Avro, JSON Schema and Protobuf serdes share one shape, with the registry client's subject strategies, rules and caching as you configured them; the suite exercises Avro and JSON Schema against the client's in-memory registry. Give it the serializer, the deserializer, or both. The topic the registry sees is the original topic of the message, so the same subject serves orders and its retry topics, and redrive() reads the DLQ under the original topic too; a consumer subscribed to the DLQ topic directly asks the registry about that topic's own subject.

What it adds is the classification the pipelines need. A registry that is unavailable (an HTTP 5xx, 429, 401 or 403, a connection or TLS error, a bearer token that could not be obtained) is a transient fault: the produce rejects with a retryable AdapterError, a consumer walks the message down the retry ladder and decodes it once the registry is back, and a redrive() stops on it, uncommitted, to be run again. A value the schema refuses, a subject that does not exist, or bytes no schema describes are a SerializationError: deterministic, so straight to the DLQ.

Headers

Headers are strings, both ways. The automatic ones use the x- prefix (shared with the RabbitMQ sibling library); change it per harbor:

import { createHarbor } from 'kafka-harbor'
import { confluentAdapter } from 'kafka-harbor/adapters/confluent'

const harbor = createHarbor({
  clientId: 'orders-service',
  brokers,
  adapter: confluentAdapter(),
  headers: {
    prefix: '',                                   // 'correlation-id', 'retry-count', ...
    correlationId: () => asyncLocalStorage.getStore()?.requestId ?? randomUUID()
  }
})
harbor.headerNames.retryCount   // the names in effect, for code that reads them

Events

harbor
  .on('connected', ({ adapter }) => {})
  .on('disconnected', ({ adapter }) => {})
  .on('messageProcessed', ({ topic, partition, offset, groupId, durationMs, correlationId, replayed, batch }) => {})   // replayed: the idempotency engine answered; batch: its size, for a message handled in a batch
  .on('batchProcessed', ({ topic, partition, groupId, size, durationMs, outcome }) => {})   // one per subscribeBatch handler run
  .on('messageFailed', ({ topic, offset, error, outcome }) => {})   // outcome: 'retry' | 'dead-letter' | 'abort' | 'crash'
  .on('messageRetried', ({ topic, retryTopic, level, attempt, error }) => {})
  .on('messageDeadLettered', ({ topic, dlqTopic, attempts, error }) => alert(`${topic}: ${attempts} attempts, now in ${dlqTopic}`))
  .on('messageRedriven', ({ from, to, offset }) => {})
  .on('messageProduced', ({ topic, kind, records, durationMs }) => {}) // one produce call acknowledged; kind: 'send' | 'retry' | 'dead-letter' | 'redrive'
  .on('consumerStopped', ({ groupId, reason }) => {})               // reason: 'shutdown' | 'abort' | 'crash'
  .on('error', ({ error, scope, groupId, topic }) => {})            // scope: 'consumer' | 'producer' | 'adapter'

messageDeadLettered fires after the DLQ produce was acknowledged, never on the attempt. error with the adapter scope covers the client: a connection that failed (once per attempt, however many calls were waiting on it), a fetch loop that reported a failure, a lag collection that could not read the offsets. A listener that throws is reported to the logger and does not affect processing.

Errors

Every error carries a stable code; message text is documentation, not contract. Branch on code or use the guards, never on instanceof: an application can end up with the ESM and CJS builds of this package in one process, and class identity does not cross that line.

| Class | code | When | |---|---|---| | ConfigError | CONFIG_INVALID | an option is invalid; thrown at construction, naming the option | | SerializationError | SERIALIZATION | a value cannot be encoded or decoded faithfully | | AbortProcessingError | ABORT_PROCESSING | built by harbor.abort(); stops the consumer without committing | | TopicMissingError | TOPIC_MISSING | a retry or DLQ topic does not exist and autoCreateTopics is off | | AdapterError | ADAPTER | the client reported a failure | | ClosedError | CLOSED | the harbor is shutting down or closed | | ShutdownTimeoutError | SHUTDOWN_TIMEOUT | handlers were abandoned by shutdown; inFlight says how many | | BatchFailedError | BATCH_FAILED | thrown by a batch handler to fail only the messages it names; cause decides retry or DLQ for them |

Every error has retryable. Throw any error with retryable: false from a handler and it goes straight to the DLQ; breakwater's errors and the RabbitMQ sibling's RetryableError follow the same convention.

Adapters

import { confluentAdapter } from 'kafka-harbor/adapters/confluent'

confluentAdapter({
  global: { 'socket.keepalive.enable': true },       // librdkafka properties, every client
  producer: { 'linger.ms': 5 },
  consumer: { 'fetch.min.bytes': 1024 },
  adminTimeoutMs: 30_000
})

The Confluent adapter sets acks=all and enable.idempotence=true on the producer, enable.auto.commit=false on consumers, and loads the client module on first connect, so importing the adapter never touches the native binding. The three properties the offset policy depends on (enable.auto.commit, enable.auto.offset.store, auto.offset.reset) cannot be overridden through the passthrough; fromBeginning drives the reset policy. Client failures are retryable unless their code is definitive (authorization, oversized record, invalid argument), regardless of the client's own retriable flag, which only describes transactions.

import { platformaticAdapter } from 'kafka-harbor/adapters/platformatic'

platformaticAdapter({
  global: { connectTimeout: 5_000 },                  // options of every client the adapter opens
  producer: { compression: 'gzip' },
  consumer: { sessionTimeout: 30_000, maxWaitTime: 500 },
  bufferedMessages: 1000,                             // messages held per consumption before the stream is left alone
  reconnectDelayMs: 1000                              // the wait before a stream the client gave up on is opened again
})

The platformatic adapter runs on @platformatic/kafka, pure TypeScript, no binary to download or compile. It pins acks=-1 and an idempotent producer with a bounded number of client retries (the client would retry forever on its own), and autocommit=false with mode: 'committed' on consumers, fromBeginning choosing the fallback. The client delivers every partition of a consumption through one stream; the adapter queues per partition and runs one handler per partition up to concurrency, reading the stream only while fewer than bufferedMessages are waiting. Heartbeats run on the client's own timer, so a long handler never gets the member kicked out, and maxProcessingTime only bounds the harbor's own wait. Partitions are assigned among the members that subscribed to each topic (partitionAssignerBySubscription, replaceable through consumer.partitionAssigner): the client's own assigner spreads every topic over every member, which strands partitions when members of a group subscribe to different topics, and one member per retry level is how the harbor lays a group out. Each worker slot takes one message and moves to the next partition in line, so a partition with a long queue never starves the others. Two differences from the Confluent adapter are worth knowing. A rebalance cannot be held back: a handler still running on a partition this member lost finishes and commits, and the new owner may already be past that offset, which costs a duplicate and never a message; what was queued for a lost partition is dropped, and a partition that comes back is read from a fresh stream, never from what the old one had fetched. Client answers Kafka marks as not retriable, authentication failures, refused arguments and a codec or API the client lacks are definitive; transport failures and timeouts are retryable.

Writing your own adapter means implementing ClientAdapter and running runAdapterContract from kafka-harbor/testing against your backend. The contract is small on purpose: connect, disconnect, produce with acknowledgment, consume with per-partition ordering and a settled-promise gate, commit, stop, optional pause/resume, two admin calls, and two optional offset calls that lag() is computed from. Everything else lives in the core.

NestJS and decorators

import { Injectable, Module } from '@nestjs/common'
import type { HandlerContext, Message } from 'kafka-harbor'
import { confluentAdapter } from 'kafka-harbor/adapters/confluent'
import { KafkaBatchListener, KafkaConsumer, KafkaListener, KafkaRetry } from 'kafka-harbor/decorators'
import { KafkaHarborModule } from 'kafka-harbor/nestjs'

@Injectable()
@KafkaConsumer({ groupId: 'orders-service', autoCreateTopics: true })
@KafkaRetry({ levels: [{ delay: '30s' }, { delay: '5m' }] })
export class OrdersListener {
  @KafkaListener<Order>('orders')
  async onOrder (message: Message<Order>, ctx: HandlerContext): Promise<void> { await fulfill(message.value) }

  @KafkaBatchListener<Order>('invoices', { size: 100, maxWait: '1s' })
  async onInvoices (messages: Array<Message<Order>>): Promise<void> { await bulkInsert(messages.map((message) => message.value)) }
}

@Module({
  imports: [KafkaHarborModule.forRoot({ clientId: 'orders-service', brokers, adapter: confluentAdapter() })],
  providers: [OrdersListener]
})
export class AppModule {}

Two entry points. kafka-harbor/decorators holds the decorators and bindListeners(), and depends on nothing but the core. kafka-harbor/nestjs holds the module, and needs @nestjs/common and @nestjs/core (optional peer dependencies of the package). KafkaHarborModule.forRoot(config) takes a HarborConfig, builds the Harbor (injectable under KAFKA_HARBOR; the module is global by default), discovers every singleton provider and controller with decorated methods when the application bootstraps, binds and starts one consumer per class, and shuts the harbor down with the application (app.close(), or enableShutdownHooks() for signals). If one consumer fails to start, the harbor is shut down before the failure propagates, so a failed bootstrap leaves nothing consuming; a decorated class that is not a singleton (request or transient scope) is refused. forRootAsync({ imports, inject, useFactory }) is there for a configuration that comes from other providers. One module per application: the discovery covers every module.

• @KafkaConsumer(options) on the class gives the consumer its group and any other ConsumerOptions; @KafkaRetry(retry) is a shorthand for the retry part.
• @KafkaListener(topic, options?) and @KafkaBatchListener(topic, options?) on methods are subscribe() and subscribeBatch(); the method is called on the provider instance with the same arguments a handler gets.
• @DLQHandler(originalTopic) subscribes the method to that topic's DLQ, named by the consumer's dlq.topicNaming. A consumer never consumes the DLQ it fills, so it goes in a class of its own, with its own group.

The decorators keep their own metadata and accept both decorator dialects: TypeScript's standard decorators and the experimentalDecorators NestJS applications compile with. A method a subclass overrides and decorates again is one listener, the nearest declaration winning; of two class decorators the upper one wins where they overlap. Without NestJS, bindListeners(harbor, instance, overrides?) returns the consumer a decorated instance declares, not started, so any framework or none can host the same classes.

Testing your handlers

import { createHarbor } from 'kafka-harbor'
import { memoryAdapter } from 'kafka-harbor/testing'

const adapter = memoryAdapter()
const harbor = createHarbor({ clientId: 'test', brokers: ['memory'], adapter })

const consumer = harbor.consumer({ groupId: 'g', fromBeginning: true, autoCreateTopics: true, retry: { levels: [{ delay: 0 }] } })
consumer.subscribe('orders', onOrder)
await harbor.producer().send('orders', { value: { id: 1 } })
await consumer.start()
await adapter.whenDrained('g', 'orders')

adapter.messages('orders-dlq')        // what landed where
adapter.committed('g', 'orders', 0)   // '1'
adapter.calls                         // every adapter call, in order

No Docker, no broker, real pipeline: the same core code that runs in production drives an in-memory broker with topics, partitions, consumer groups and committed offsets. examples/testing-handlers.ts is a complete handler test written this way; examples/retry-dlq-flow.ts runs the whole retry and DLQ flow against the broker from docker-compose.yml.

Development

npm install
npm run hooks              # once per clone
npm test                   # unit + adapter contract on the memory adapter
npm run test:integration   # the same contract on a real Kafka, plus the retry/DLQ flow (needs Docker)
npm run lint
npm run check:types && npm run check:types:next
npm run check:dist         # build, then compile a consumer against the published declarations
npm run check:docs         # every ```ts block in the docs type-checks against src/; every anchor resolves
npm run api:check          # public API frozen by the report in etc/
npm run test:mutation:changed

See CONTRIBUTING.md for the invariants worth knowing before changing anything, and docs/ for delivery semantics and adapter authoring.

Related

• breakwater: resilience policies (retry, circuit breaker, bulkhead). kafka-harbor's produce retry runs on it.
• quayside: idempotency with pluggable storage. Its engine fits idempotency.engine as is.

License

MIT