Kafka Client

Native Apache Kafka client for JavaScript and TypeScript runtimes. Implements the Kafka binary protocol directly for producing, consuming, and administering Kafka clusters.

Features

• Zero native dependencies — pure TypeScript binary protocol implementation
• Multi-runtime — Bun, Node.js 20+, and Deno
• Producer — batching, partitioning (murmur2/round-robin/custom), retries, idempotent mode
• Consumer — group coordination, offset management, rebalance listeners, auto-commit
• Admin — topic/partition CRUD, config describe/alter
• SASL authentication — PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, with rotating credentials
• SSL/TLS — mutual TLS support via runtime socket adapters
• Serialisation — built-in JSON/string, pluggable Avro and Protobuf via Schema Registry
• Compression — gzip, snappy, lz4, zstd

Installation

bun add @qualithm/kafka-client
# or
npm install @qualithm/kafka-client

Quick Start

import { createKafka, createNodeSocketFactory } from "@qualithm/kafka-client"

const kafka = createKafka({
  config: { brokers: ["localhost:9092"], clientId: "my-app" },
  socketFactory: createNodeSocketFactory()
})
await kafka.connect()

const producer = kafka.producer()
await producer.send("my-topic", [
  { key: new TextEncoder().encode("key-1"), value: new TextEncoder().encode("hello") }
])
await producer.close()
await kafka.disconnect()

Usage

Consuming Messages

import { createKafka, createNodeSocketFactory } from "@qualithm/kafka-client"

const kafka = createKafka({
  config: { brokers: ["localhost:9092"], clientId: "my-app" },
  socketFactory: createNodeSocketFactory()
})
await kafka.connect()

const consumer = kafka.consumer({ groupId: "my-group" })
consumer.subscribe(["my-topic"])
await consumer.connect()

const records = await consumer.poll()
for (const record of records) {
  console.log(new TextDecoder().decode(record.message.value!))
}

await consumer.close()
await kafka.disconnect()

Admin Operations

const admin = kafka.admin()
await admin.createTopics({
  topics: [{ name: "new-topic", numPartitions: 3, replicationFactor: 1 }],
  timeoutMs: 30000
})
const topics = await admin.listTopics()

Runtime Adapters

// Bun
import { createBunSocketFactory } from "@qualithm/kafka-client"
const socketFactory = createBunSocketFactory()

// Node.js
import { createNodeSocketFactory } from "@qualithm/kafka-client"
const socketFactory = createNodeSocketFactory()

// Deno
import { createDenoSocketFactory } from "@qualithm/kafka-client"
const socketFactory = createDenoSocketFactory()

Rotating Credentials

Pass an authProvider instead of a static sasl block when the credential is short-lived. It is called on every connection attempt — including the reconnects the pool performs after a broker drops — so a rotated secret is adopted without restarting the client or triggering a consumer-group rebalance.

import { createKafka, createNodeSocketFactory } from "@qualithm/kafka-client"

const kafka = createKafka({
  config: {
    brokers: ["localhost:9092"],
    authProvider: async () => {
      const { username, password } = await vault.read("kafka/creds")
      return { mechanism: "SCRAM-SHA-512", username, password }
    }
  },
  socketFactory: createNodeSocketFactory()
})

The provider may be synchronous or asynchronous, and takes precedence over sasl when both are set. Cache the credential yourself if resolving it is expensive — it is resolved once per connection, not once per request.

Compression

Register compression providers before producing or consuming compressed record batches:

import {
  CompressionCodec,
  createSnappyProvider,
  registerCompressionProvider
} from "@qualithm/kafka-client"
import snappy from "snappy" // bring your own codec

registerCompressionProvider(
  CompressionCodec.SNAPPY,
  createSnappyProvider({ compressSync: snappy.compressSync, decompressSync: snappy.uncompressSync })
)

Available: gzipProvider, deflateProvider, createSnappyProvider, createLz4Provider, createZstdProvider.

Schema Registry

import { SchemaRegistry, createAvroSerde } from "@qualithm/kafka-client"

const registry = new SchemaRegistry({ url: "http://localhost:8081" })
const serde = createAvroSerde<MyType>({ registry, schema: mySchemaJson, codec: avroCodec })

// Serialize for producing
const encoded = await serde.serialize(myData, "my-topic")

// Deserialize when consuming
const decoded = await serde.deserialize(record.message.value!, "my-topic")

API Reference

Full API documentation is generated with TypeDoc:

bun run docs
# Output in docs/

Examples

See the examples/ directory for runnable examples:

Example	Description
basic-usage.ts	Core types, broker parsing, version negotiation
batch-processing.ts	Construct and encode record batches
produce-consume.ts	End-to-end produce and consume with a live broker
error-handling.ts	Error hierarchy and type narrowing

bun run examples/basic-usage.ts

Development

Prerequisites

• Bun (recommended), Node.js 20+, or Deno

Setup

bun install

Building

bun run build

Testing

bun run test              # unit tests
bun run test:integration  # integration tests (requires a running broker)
bun run test:coverage     # with coverage report

Linting & Formatting

bun run lint
bun run format
bun run typecheck

Benchmarks

bun run bench

License

Apache-2.0