Kafka in production in TypeScript: kafkajs, DLQ, Schema Registry, tuning, security

← Back to the section

If you already know what a topic, a partition, and a consumer group are, this article is about the next step: how to connect Kafka to a NestJS application and what you need to configure so that it runs reliably in production.

kafkajs: producer and eachMessage

The kafkajs library is the standard Kafka client for Node.js — it implements the Kafka protocol in pure JavaScript and gives you a convenient API. To send messages you use producer.send; to receive them you use consumer.run with an eachMessage handler.

// Send
@Injectable()
export class OrderEventPublisher {
  constructor(private readonly producer: Producer) {}

  async publish(event: OrderEvent): Promise<void> {
    await this.producer.send({
      topic: 'orders',
      messages: [{ key: event.orderId.toString(), value: JSON.stringify(event) }],
    });
  }
}

// Receive
const consumer = kafka.consumer({ groupId: 'billing-service' });
await consumer.subscribe({ topic: 'orders' });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event: OrderEvent = JSON.parse(message.value!.toString());
    // processing
  },
});

By default consumer.run processes partitions one at a time, but you can request more (partitionsConsumedConcurrently: 3) — then several partitions are processed in parallel.

Commit modes: when Kafka "knows" a message has been processed

Kafka tracks how far a consumer has progressed through the messages via an offset commit. The consumer.run settings determine when that commit happens:

autoCommit: true — offsets are committed automatically, by interval (autoCommitInterval) or by message threshold (autoCommitThreshold) (the default value).
eachMessage + autoCommit — kafkajs resolves the offset after each successfully processed message. Safer on failures, but slower.
autoCommit: false — the application calls consumer.commitOffsets() itself. Needed when it matters to commit the offset only after a successful write to the database.

await consumer.run({
  partitionsConsumedConcurrently: 3,
  autoCommit: false,
  eachMessage: async ({ topic, partition, message }) => {
    await saveToDatabase(message);
    await consumer.commitOffsets([
      { topic, partition, offset: (Number(message.offset) + 1).toString() },
    ]);
  },
});

Message headers

Every Kafka message can carry headers — (key, value) pairs in bytes, separate from the payload. This is where you put technical metadata that does not belong in the business data:

X-Correlation-ID / traceparent — for distributed tracing.
X-Event-Version — the version of the event format.
X-Source-Service — where the message came from.

// When sending
await producer.send({
  topic: 'orders',
  messages: [{
    key,
    value: JSON.stringify(event),
    headers: { 'X-Correlation-ID': correlationId },
  }],
});

// When receiving
await consumer.run({
  eachMessage: async ({ message }) => {
    const correlationId = message.headers?.['X-Correlation-ID']?.toString();
    await asyncLocalStorage.run({ correlationId }, async () => {
      // processing
    });
  },
});

Business data (orderId, amount, status) goes in the payload. Everything technical goes in the headers.

Dead Letter Queue: what to do with problematic messages

If a handler throws an exception, by default kafkajs keeps trying to process the same message over and over — the offset does not move and the consumer stalls. In production this means the whole group grinds to a halt.

The solution is a Dead Letter Queue (DLQ): after several failed attempts the message is moved to a separate topic (usually with a .DLT suffix), the offset is committed, and the main consumer keeps working.

The simple way: retries in the handler plus publishing to the DLT

await consumer.run({
  eachMessage: async ({ topic, message }) => {
    try {
      await retry(() => handle(message), { retries: 3, minTimeout: 1000 });
    } catch (e) {
      await producer.send({
        topic: 'orders.DLT',
        messages: [{
          key: message.key,
          value: message.value,
          headers: {
            'x-exception-message': (e as Error).message,
            'x-original-topic': topic,
          },
        }],
      });
    }
  },
});

Here: 3 attempts with a 1-second interval, and after the third failure the message goes to orders.DLT. The DLT message gets headers with the error text and the name of the original topic.

The non-blocking way: retry topics

kafkajs has no ready-made declarative equivalent — the same pattern is assembled by hand: a failed message is re-sent to a retry topic with an ever-increasing delay, and once the attempts are exhausted — to the DLT.

const RETRY_TOPICS = ['orders-retry-0', 'orders-retry-1', 'orders-retry-2'];

async function reroute(message: KafkaMessage, attempt: number, error: Error): Promise<void> {
  const next = attempt < RETRY_TOPICS.length ? RETRY_TOPICS[attempt] : 'orders-dlt';
  await producer.send({
    topic: next,
    messages: [{
      key: message.key,
      value: message.value,
      headers: {
        'x-attempt': String(attempt + 1),
        'x-retry-at': String(Date.now() + 1000 * 2 ** attempt),
        'x-exception-message': error.message,
      },
    }],
  });
}

The consumer of the retry topics waits until x-retry-at before processing. Messages "cool off" in the orders-retry-0, orders-retry-1, orders-retry-2 topics without blocking the main flow.

When to use which

Temporary problem (database down, external service not responding) → retry topics with an increasing delay.
Permanent problem (invalid message, wrong format) → no retries, straight to the DLQ.
Need to distinguish error types → check the exception type in catch: for example, a SyntaxError from JSON.parse goes straight to the DLQ, everything else gets retried.

Schema Registry: how not to break your neighbors when the schema changes

When the producer and the consumer are different services from different teams, any change to the event structure can potentially break the consumer. Schema Registry solves this with a centralized schema store that checks compatibility.

How it works:

The producer registers the schema in Schema Registry and gets a numeric identifier (schema_id).
Into each message it writes 4 bytes with that identifier, followed by the compressed payload.
The consumer reads the schema_id, downloads the schema from Schema Registry once (then caches it), and deserializes the payload.

Avro, Protobuf, or JSON Schema

The standard in the Kafka ecosystem is Avro: a compact binary format with good support for schema evolution; in Node.js you work with it via @kafkajs/confluent-schema-registry. Protobuf is chosen by teams with gRPC infrastructure. JSON Schema is human-readable but takes several times more space — it is good for debugging.

Compatibility modes

Schema Registry checks whether a new version of the schema will break consumers that are already running:

BACKWARD (the default) — a new consumer can read old messages. You can remove optional fields and add optional ones with a default value. You cannot add required fields.
FORWARD — an old consumer can read new messages. The mirror image of BACKWARD.
FULL — both modes at once. The strictest.
NONE — no checks. Only for special cases.

Rule of thumb: BACKWARD for topics where there are many consumers and a single producer — that is the typical event bus.

Consumer lag: why the consumer falls behind

Lag is the difference between the last message in a partition and the one the consumer has reached. If lag is growing, the consumer cannot keep up with the producer. This is the main health metric for a consumer in production.

Typical causes:

Slow processing — each message makes a synchronous call to a database or an external service. Fix: reduce the batch size (maxBytesPerPartition), parallelize processing, switch to batch requests.
Too few partitions per group — increase partitionsConsumedConcurrently or add service instances (but no more than the number of partitions).
sessionTimeout expires — if processing takes longer than the session timeout (30 seconds by default) without a heartbeat, Kafka considers the consumer dead and triggers a partition rebalance. Fix: call heartbeat() during long processing or raise sessionTimeout.
Event loop blocking / GC pauses — the process froze for 30 seconds and lag grew by thousands of messages.

Check the lag manually:

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group billing-service --describe
# TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# orders  0          124530          124530          0
# orders  1          124450          124530          80   ← falling behind

Metric to monitor: kafka_consumergroup_lag. Alert: lag is above N and has been growing for N minutes in a row.

Performance tuning

Kafka works well out of the box, but there are a few parameters worth knowing.

Producer

Parameter	Default value	When to change
Batch size	1 `send` call = 1 batch	Under high write volume, accumulate messages and send them as a `messages` array in a single call — kafkajs does not buffer between calls, the application forms the batch.
`compression`	`None`	Enable `CompressionTypes.ZSTD` or `LZ4` (plugged in via external codecs). On JSON messages it saves 3–5x on traffic and disk space.
`acks`	`-1` (all replicas)	Keep `-1` for business events; `1` or `0` only for metrics and telemetry where loss is acceptable.

Consumer

Parameter	Default value	When to change
`minBytes`	1 byte	Raise to 50–500 KB. The consumer waits for data to accumulate — less load on the broker.
`maxBytesPerPartition`	1 MB	Lower it if each message requires long processing — batches become smaller.
`sessionTimeout`	30 seconds	Raise it (and call `heartbeat()` in the handler) if processing a batch really does take that long.

The best compression algorithm for most cases is zstd (Kafka 2.1+): it compresses almost as well as gzip and runs almost as fast as lz4.

Security: SASL, SSL, ACL

By default Kafka listens on an open port with no authentication — you never leave it that way in production. Three layers of protection:

SSL/TLS — connection encryption. Certificates on the brokers and clients.
SASL — authentication. SASL/PLAIN — username and password (only over SSL). SASL/SCRAM-SHA-256 — safer: challenge-response, the password is not sent over the network. SASL/OAUTHBEARER — OAuth2 tokens, for integration with an identity provider.
ACL — authorization: who is allowed to read from and write to which topic. Managed via kafka-acls.sh.

Typical setup: one SASL user per service, with ACLs restricting its access strictly to the topics it needs.

const kafka = new Kafka({
  clientId: 'billing-service',
  brokers: ['kafka:9093'],
  ssl: {
    ca: [fs.readFileSync('/etc/kafka/ca.pem', 'utf-8')],
  },
  sasl: {
    mechanism: 'scram-sha-256',
    username: 'billing-service',
    password: process.env.KAFKA_PASSWORD!,
  },
});

KRaft: Kafka without ZooKeeper

Kafka used to require a separate ZooKeeper cluster to store metadata: which topics exist, who is the partition leader, which ACLs are in effect. ZooKeeper is a separate system that has to be deployed, monitored, and fixed on its own.

As of version 3.3 (2022) there is KRaft — a native Raft-based protocol running inside the brokers themselves. ZooKeeper is no longer needed. In Kafka 4.x ZooKeeper support has been removed entirely — all new clusters run on KRaft only.

For a developer this is almost invisible: bootstrap-servers and client behavior have not changed. The only thing that changes is how the ops team deploys and maintains the cluster.

In short

producer.send sends, consumer.run with eachMessage receives. Per-partition parallelism — partitionsConsumedConcurrently.
The commit mode controls when the offset is committed: autoCommit by interval/threshold (default) or a manual commitOffsets after the database write.
Message headers are the place for technical metadata (trace-id, schema version). Business data goes in the payload.
The DLQ saves you from a consumer hanging on a problematic message. Retry topics with an increasing delay are the way to retry without blocking the main flow.
Schema Registry stores schemas and checks compatibility on changes. BACKWARD mode — a new consumer reads old messages.
Consumer lag is the main health metric for a consumer. Growing → the consumer cannot keep up.
Producer tuning: batch the messages in a single send, compression=zstd. Consumer tuning: minBytes, maxBytesPerPartition, sessionTimeout.
Security: SSL for encryption, SASL for authentication, ACL for per-topic authorization.
KRaft: as of Kafka 3.3+ ZooKeeper is not needed; in Kafka 4.x ZooKeeper support has been removed entirely.