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

Producing Messages

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()

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.

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 { registerCompressionProvider, createSnappyProvider } from "@qualithm/kafka-client"
import snappy from "snappy" // bring your own codec

registerCompressionProvider(createSnappyProvider(snappy))

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

Schema Registry

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

const registry = new SchemaRegistry({ baseUrl: "http://localhost:8081" })
const serde = createAvroSerde<MyType>({ registry, subject: "my-topic-value", codec: avroCodec })

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

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

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 | Connect, produce, and consume | | batch-processing.ts | Batch produce and consume | | produce-consume.ts | End-to-end produce/consume flow | | error-handling.ts | Error handling patterns |

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

Publishing

The package is automatically published to NPM when CI passes on main. Update the version in package.json before merging to trigger a new release.

License

Apache-2.0