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
• 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
Copy

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

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

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

Admin Operations

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

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

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

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!)
Copy

API Reference

Full API documentation is generated with TypeDoc:

bun run docs
# Output in docs/
Copy

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
Copy

Development

Prerequisites

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

Setup

bun install
Copy

Building

bun run build
Copy

Testing

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

Linting & Formatting

bun run lint
bun run format
bun run typecheck
Copy

Benchmarks

bun run bench
Copy

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.

Licence

Apache-2.0