kafka-harbor: the Kafka layer every Node.js team rebuilds by hand, done once

I have written the same five hundred lines on top of a Kafka client more times than I want to admit. Different companies, different clients, same code: retry with a delay, a dead-letter topic, commit only after the handler succeeded, shut down without losing the message in flight. Every time it was "just a small wrapper", and every time it grew into the part of the service nobody wanted to touch.

So I wrote it once, properly, and published it. kafka-harbor hit 1.0.0 this week. This post is about why it exists and what it does that the clients don't.

The clients stop at the protocol

Every Kafka client for Node.js gives you a producer, a consumer, and good luck. That is not a complaint; it is their job. @confluentinc/kafka-javascript, kafkajs, @platformatic/kafka: they all speak the wire protocol well, and they all stop there.

Your service needs more than the protocol. It needs an answer to questions like:

• The handler threw. Now what? Retry? How many times? With what delay? Where does the message go when you give up?
• The process got a SIGTERM with a handler halfway through a database write. Does the offset get committed? Does the message get redelivered? Does anyone find out?
• The retry produce failed because the broker was away for a second. Did the offset get committed anyway, and the message vanish?

If you are on the JVM, Spring Kafka answers all of that for you. On Node.js, you answer it yourself, in every project, slightly differently each time.

What kafka-harbor is

It is not a client. It runs on top of one, through a small ClientAdapter interface. Two adapters ship with it: one wraps Confluent's client (librdkafka underneath), the other wraps @platformatic/kafka, pure TypeScript, no native binding. The core never sees a client type, and the same contract test suite runs against both, so switching clients is a one-line change that does not touch a single handler.

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)
  ctx.logger.info(`order ${message.key} done on attempt ${ctx.attempt}`)
})

await consumer.start()
harbor.enableSignalHandlers()

That is the whole setup. If fulfill throws, the message goes to orders-retry-1, waits five seconds, runs again. Fails again: orders-retry-2, one minute. Then ten. Then orders-dlq. On SIGTERM, running handlers finish, their offsets commit, the consumer leaves the group, the client disconnects.

The design principle behind every decision in the library: 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.

The parts I cared most about

Retry that does not block the partition

The naive retry is sleep(delay); tryAgain() inside the handler. It blocks the partition, it blocks the worker, and if the delay is longer than the poll interval the group kicks you out.

kafka-harbor re-produces the failed message, bytes untouched, to a retry topic, and the same consumer group consumes that topic. A message that is not due yet is parked: its partition is paused through the adapter, the message waits in memory, and the client keeps polling, so the group never evicts the member. Nothing is committed while parked, so a crash or a rebalance leaves the message exactly where it was.

Each retry level is a group member of its own. A message sleeping out a ten-minute delay on orders-retry-3 never holds a worker that orders is waiting for.

Every hop rewrites a small block of tracking headers: x-retry-count, x-original-topic, x-first-failure-at, x-last-error. They come from the network, so they are validated before use; a corrupt count never turns into 0 or NaN.

A DLQ you can actually drain

The dead-letter topic gets the original bytes plus the failure trail in the headers. And when the bug is fixed, you put the messages back:

const result = await harbor.redrive({
  from: 'orders-dlq',
  max: 500,
  filter: (message) => message.retry?.lastError !== 'ValidationError: bad sku'
})
// { from: 'orders-dlq', reprocessed: 498, skipped: 2 }

Each message goes back to its own original topic with a fresh retry ladder. The DLQ offset is committed only after the broker acknowledged the re-produce, so an interrupted redrive resumes where it stopped.

Commit only after the broker said yes

This is the one that bit me the hardest in hand-written wrappers. The handler fails, you produce to the retry topic, you commit the offset. If the produce silently failed, you just committed past a message that is now nowhere.

In kafka-harbor the offset is committed only after the retry or DLQ produce was acknowledged. A produce that cannot be acknowledged stops the consumer instead of committing. Loud, not lost.

And when reprocessing is the right answer (a schema the handler cannot read yet, a dependency that is down and must not be skipped), the handler says so:

consumer.subscribe('orders', async (message) => {
  if (!schemaReady()) throw harbor.abort(new Error('schema not migrated yet'))
  await fulfill(message.value)
})

abort stops the consumer without committing. The message is redelivered after a restart. No retry topic, no DLQ, no guessing.

A shutdown that tells the truth

await harbor.shutdown('30s')

Consumers stop taking new messages. Running handlers get up to the timeout to finish; the ones that finish commit their offsets on the way out, and what they produce on the way (a send, a transaction, a retry hop) still goes through. Handlers still running when the timeout elapses are abandoned: their ctx.signal aborts, their offsets are not committed, and shutdown() rejects with a ShutdownTimeoutError that says how many. At-least-once, said out loud, instead of a disconnect() that returns and leaves you wondering.

Tests without Docker

This is the feature I use the most, day to day:

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'

The same core code that runs in production drives an in-memory broker with topics, partitions, consumer groups and committed offsets. A handler test runs in milliseconds and exercises the real retry and DLQ flow, not a mock of it.

The rest, briefly

• Transactions. harbor.transaction() makes what you produced land together or not at all; inside a handler, ctx.transaction() commits the consumed offset along with it. Exactly-once from an input topic to output topics. If another instance takes your transactional id, the fenced one stops instead of dead-lettering what the other one is processing.
• Circuit breaker per topic. When the dependency behind a handler goes down, the partition is held instead of flooding the retry topics with work that will fail again. Built on breakwater.
• Idempotency. Run every handler once per key through a pluggable engine; a redelivery replays the first outcome and commits without running the handler again.
• Serialization that refuses to lie. The default JSON serializer rejects Map, undefined and NaN instead of silently flattening them. Pluggable per topic, with a Schema Registry entry point.
• Observability. kafka-harbor/prometheus and kafka-harbor/otel give you metrics and traces where one message's attempts, its dead-lettering and its redrive form one chain. Typed events (messageRetried, messageDeadLettered, consumerStopped, ...) carry a correlation id.
• NestJS. Decorators and a module, so a consumer is a class with @KafkaListener('orders') on a method, and the application's shutdown drains it in the right order.
• Safe defaults. Idempotent producer and acks=all are set by the adapter and cannot be overridden by accident.

What it does not do

I would rather you know this before you install it.

• At-least-once only. Duplicates are possible after a crash between handler and commit. The docs list every case. Exactly-once effects come from idempotency, not from the transport.
• Retry breaks ordering. A message that goes through a retry topic runs after later messages on the original topic. If order matters more than progress, harbor.abort() blocks the partition instead.
• Transactions cover Kafka only. A database write inside tx is not rolled back by an abort.
• A parked retry holds one message per partition in memory. Long delays on many partitions mean that many messages waiting in the process. Nothing is lost if it dies, but it is memory.

Why 1.0 now

The API has been running against a real broker end to end, on both adapters, for a while. The behaviour that matters is documented as guarantees, not as happy paths: what is committed when, what is redelivered after what, what a shutdown abandons. The public surface is frozen by an API report per entry point, so a change to it fails CI before it ships. From here on, the version number means what semver says it means.

Try it

npm install kafka-harbor @confluentinc/kafka-javascript
# or, pure TypeScript, no native build:
npm install kafka-harbor @platformatic/kafka

Node.js 22 or newer. ESM and CommonJS, declarations for both.

The README on GitHub has every option, the delivery semantics document lists every duplicate you can expect, and there is a guide for writing an adapter for a client I did not cover. If you have rebuilt this layer yourself and your version handles a case mine does not, open an issue. That is exactly the kind of feedback I am after.