---
url: https://chrisrecalis.github.io/kafkats/guide.md
---
# Getting Started

kafkats is a pure-protocol Kafka client and streams library for TypeScript. Unlike other Node.js Kafka clients that wrap librdkafka, kafkats implements the Kafka wire protocol directly in TypeScript.

## Why kafkats?

* **Pure TypeScript** - No native dependencies, works everywhere Node.js runs
* **Type-safe** - Full TypeScript support with comprehensive types
* **High performance** - Optimized for throughput with zero-copy operations
* **Modern** - ESM-first, async/await, tree-shakeable
* **Complete** - Producer, consumer, transactions, and stream processing

## Packages

| Package                                       | Description                                  |
| --------------------------------------------- | -------------------------------------------- |
| [@kafkats/client](/client/)                   | Core Kafka client with producer and consumer |
| [@kafkats/flow](/flow/)                       | Kafka Streams-like DSL for stream processing |
| [@kafkats/flow-codec-zod](/flow-codec-zod/)   | Zod schema validation codecs                 |
| [@kafkats/flow-state-lmdb](/flow-state-lmdb/) | LMDB-backed persistent state stores          |

## Next Steps

1. [Install the packages](/guide/installation)
2. [Follow the quick start tutorial](/guide/quick-start)
3. [Learn core concepts](/guide/concepts)

---

---
url: https://chrisrecalis.github.io/kafkats/guide/installation.md
---
# Installation

## Requirements

* Node.js 18 or later
* A running Kafka broker (for integration)

## Installing the Client

Install the core client package:

::: code-group

```bash [pnpm]
pnpm add @kafkats/client
```

```bash [npm]
npm install @kafkats/client
```

```bash [yarn]
yarn add @kafkats/client
```

:::

## Installing Flow (Stream Processing)

For Kafka Streams-like processing, install the flow package:

::: code-group

```bash [pnpm]
pnpm add @kafkats/flow
```

```bash [npm]
npm install @kafkats/flow
```

```bash [yarn]
yarn add @kafkats/flow
```

:::

## Optional Packages

### Native CRC32C (Recommended)

For maximum producer/consumer throughput, install the optional native CRC32C implementation:

::: code-group

```bash [pnpm]
pnpm add @node-rs/crc32
```

```bash [npm]
npm install @node-rs/crc32
```

```bash [yarn]
yarn add @node-rs/crc32
```

:::

With native CRC32C enabled, kafkats can exceed the throughput of other popular Kafka clients.

If not installed, kafkats falls back to a pure TypeScript CRC32C implementation.

### Zod Codec

For schema validation with Zod:

```bash
pnpm add @kafkats/flow-codec-zod zod
```

### LMDB State Store

For persistent state in stream processing:

```bash
pnpm add @kafkats/flow-state-lmdb
```

::: warning Native Dependencies
The LMDB package includes native bindings. Make sure you have the necessary build tools installed on your system.
:::

## TypeScript Configuration

kafkats is written in TypeScript and includes type definitions. For the best experience, ensure your `tsconfig.json` includes:

```json
{
	"compilerOptions": {
		"module": "ESNext",
		"moduleResolution": "bundler",
		"strict": true,
		"esModuleInterop": true
	}
}
```

## Verifying Installation

Create a simple test file to verify the installation:

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'test',
	brokers: ['localhost:9092'],
})

console.log('kafkats installed successfully!')
```

Run it:

```bash
npx tsx test.ts
```

---

---
url: https://chrisrecalis.github.io/kafkats/guide/quick-start.md
---
# Quick Start

This guide will help you send and receive your first messages with kafkats.

## Prerequisites

Make sure you have:

* [Installed @kafkats/client](/guide/installation)
* A running Kafka broker (e.g., via Docker)

Start Kafka locally with Docker:

```bash
docker run -d --name kafka \
  -p 9092:9092 \
  -e KAFKA_CFG_NODE_ID=0 \
  -e KAFKA_CFG_PROCESS_ROLES=controller,broker \
  -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@localhost:9093 \
  -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  bitnami/kafka:latest
```

## Creating a Client

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})
```

## Producing Messages

```typescript
const producer = client.producer()

// Send a single message
await producer.send('my-topic', [{ value: 'Hello, Kafka!' }])

// Send multiple messages with keys
await producer.send('my-topic', [
	{ key: 'user-1', value: JSON.stringify({ action: 'login' }) },
	{ key: 'user-2', value: JSON.stringify({ action: 'signup' }) },
])

// Don't forget to close when done
await producer.disconnect()
```

## Consuming Messages

```typescript
const consumer = client.consumer({
	groupId: 'my-group',
	autoOffsetReset: 'earliest',
})

// Process messages one at a time
await consumer.runEach('my-topic', async (message, ctx) => {
	console.log({
		topic: ctx.topic,
		partition: ctx.partition,
		offset: ctx.offset,
		key: message.key?.toString(),
		value: message.value?.toString(),
	})
})
```

## Complete Example

Here's a complete example that produces and consumes messages:

```typescript
import { KafkaClient } from '@kafkats/client'

async function main() {
	const client = new KafkaClient({
		clientId: 'quickstart-app',
		brokers: ['localhost:9092'],
	})

	// Create producer and send messages
	const producer = client.producer()

	await producer.send('quickstart', [
		{ key: 'greeting', value: 'Hello from kafkats!' },
		{ key: 'farewell', value: 'Goodbye from kafkats!' },
	])

	console.log('Messages sent!')
	await producer.disconnect()

	// Create consumer and read messages
	const consumer = client.consumer({
		groupId: 'quickstart-group',
		autoOffsetReset: 'earliest',
	})

	console.log('Waiting for messages...')

	let seen = 0
	await consumer.runEach('quickstart', async message => {
		console.log(`Received: ${message.key?.toString()} = ${message.value?.toString()}`)
		if (++seen >= 2) consumer.stop()
	})
}

main().catch(console.error)
```

## Next Steps

* Learn about [core concepts](/guide/concepts)
* Explore the [Producer API](/client/producer)
* Explore the [Consumer API](/client/consumer)
* Try [stream processing with Flow](/flow/)

---

---
url: https://chrisrecalis.github.io/kafkats/guide/concepts.md
---
# Core Concepts

Understanding these core concepts will help you use kafkats effectively.

## Topics, Partitions, and Ordering

A **topic** is a named log of records. Topics are divided into **partitions**, which are ordered, immutable sequences of records.

```typescript
// Produce records to a topic
await producer.send('orders', [{ key: 'order-123', value: orderJson }])
```

Key points:

* **Ordering is per partition** (not per topic).
* **Parallelism comes from partitions** (more partitions → more consumers can work in parallel).

| Concept   | What it means in practice                       |
| --------- | ----------------------------------------------- |
| Topic     | A named log of records (often one “event type”) |
| Partition | The unit of ordering and parallelism            |
| Record    | A key/value payload plus headers and timestamp  |

## Keys and Partitioning

Message **keys** determine which partition a record goes to. Records with the same key always go to the same partition, which preserves order for that key.

```typescript
await producer.send('events', [
	{ key: 'user-1', value: event1 },
	{ key: 'user-1', value: event2 }, // same partition as above
	{ key: 'user-2', value: event3 }, // potentially different partition
])
```

## Brokers, Leaders, and Replication

A Kafka **cluster** is made of **brokers**. Each partition has a **leader** broker and zero or more replica brokers. The leader handles reads/writes; replicas follow the leader.

This matters when producing:

| Producer `acks` | Meaning                              | Typical use                         |
| --------------- | ------------------------------------ | ----------------------------------- |
| `'none'`        | Don't wait for broker acknowledgment | Fire-and-forget logs (risk of loss) |
| `'leader'`      | Wait for the leader to write         | Lower latency, less durable         |
| `'all'`         | Wait for all in-sync replicas        | Most durable (recommended default)  |

## Producer Batching and Queueing

In kafkats, `producer.send()` is **queue-based**: records are appended to an in-memory accumulator and flushed as partition batches.

| Setting         | What it controls                                                        |
| --------------- | ----------------------------------------------------------------------- |
| `lingerMs`      | Time-based batching: how long to wait before flushing a partition batch |
| `maxBatchBytes` | Size-based batching: flush when the batch reaches this size             |
| `compression`   | Compression applied to record batches                                   |

See [Producer API](/client/producer) for details.

## Consumer Groups and Rebalances

**Consumer groups** allow multiple consumers to share the work of processing a topic. Each partition is assigned to exactly one consumer in the group at a time.

When consumers join/leave, Kafka performs a **rebalance** to reassign partitions.

| Assignment strategy    | Description                                                        |
| ---------------------- | ------------------------------------------------------------------ |
| `'cooperative-sticky'` | Incremental rebalancing (Kafka 2.4+), minimizes movement (default) |
| `'sticky'`             | Eager rebalance with minimized movement                            |
| `'range'`              | Simple per-topic assignment                                        |

## Offsets and Offset Resets

An **offset** is a monotonically increasing position within a partition. Kafka stores committed offsets (per group) in `__consumer_offsets`.

When a group has no committed offset (new group, offsets expired), `autoOffsetReset` decides what to do:

| Value        | Behavior                                 |
| ------------ | ---------------------------------------- |
| `'earliest'` | Start from the earliest available offset |
| `'latest'`   | Start from the end (new records only)    |
| `'none'`     | Fail if no committed offset exists       |

## Delivery Semantics

Kafka's durability is a property of the log, but what your application observes depends on how you produce, consume, and commit offsets.

| Semantics     | What you get                               | Typical approach                                              |
| ------------- | ------------------------------------------ | ------------------------------------------------------------- |
| At-most-once  | No duplicates, possible loss               | Commit before processing (rare)                               |
| At-least-once | No loss, possible duplicates               | Process → commit (common default)                             |
| Exactly-once  | No loss, no duplicates (within a topology) | Transactions + `read_committed` (use Flow for end-to-end EOS) |

## Codecs and Typed Topics

Use codecs to get type-safe key/value encode/decode on both producer and consumer.

```typescript
import { topic, string, json } from '@kafkats/client'

const userTopic = topic('users', {
	key: string(),
	value: json<{ id: string; name: string }>(),
})

await producer.send(userTopic, [{ key: 'user-1', value: { id: 'user-1', name: 'Alice' } }])
```

## Transactions and Isolation

Transactions let you write to multiple partitions/topics atomically.

```typescript
const producer = client.producer({
	transactionalId: 'my-transaction',
	acks: 'all',
})

await producer.transaction(async txn => {
	await txn.send('output', [{ value: 'processed' }])
})
```

Consumers can control whether they see uncommitted transactional data:

| `isolationLevel`     | What you see                                       |
| -------------------- | -------------------------------------------------- |
| `'read_committed'`   | Only committed transactional records (recommended) |
| `'read_uncommitted'` | All records, including uncommitted                 |

## Stream Processing Concepts

### KStream

A **KStream** represents an unbounded stream of records. Each record is an independent event.

```typescript
import { flow } from '@kafkats/flow'

const app = flow({ applicationId: 'my-app', ... })
app.stream('events')
  .filter((key, value) => value.type === 'click')
  .mapValues(value => ({ ...value, processed: true }))
  .to('processed-events')
```

### KTable

A **KTable** represents a changelog stream, where each key has a latest value. It's like a continuously-updated table.

```typescript
app.table('users') // Latest value for each user ID
	.mapValues(user => user.name)
	.to('user-names')
```

### Windowing

**Windowing** groups stream records by time for aggregations.

```typescript
import { TimeWindows } from '@kafkats/flow'

app.stream('clicks')
	.groupByKey()
	.windowedBy(TimeWindows.of('5m')) // 5-minute windows
	.count()
```

## Next Steps

* [Producer API](/client/producer) - Sending messages
* [Consumer API](/client/consumer) - Receiving messages
* [Flow Streams](/flow/streams) - Stream processing

---

---
url: https://chrisrecalis.github.io/kafkats/client.md
---
# @kafkats/client

The core Kafka client package providing producer, consumer, and low-level protocol access.

## Features

* **Pure Protocol** - Direct Kafka wire protocol implementation, no native dependencies
* **Type-Safe** - Full TypeScript support with comprehensive types
* **High Performance** - Optimized batching, zero-copy operations
* **SASL Authentication** - PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER
* **Transactions** - Full exactly-once semantics support
* **Compression** - gzip, snappy, lz4, zstd

## Installation

```bash
pnpm add @kafkats/client
```

## Quick Example

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})

// Create a producer
const producer = client.producer({
	acks: 'all',
	compression: 'snappy',
})

await producer.send('events', [{ key: 'user-1', value: JSON.stringify({ action: 'click' }) }])

// Create a consumer
const consumer = client.consumer({ groupId: 'my-group' })

await consumer.runEach('events', async message => {
	console.log(message.value.toString())
})
```

## Architecture

```
@kafkats/client
├── KafkaClient      # Main entry point
├── Producer         # Message production with batching
├── Consumer         # Consumer groups and message handling
├── Cluster          # Broker discovery and metadata
└── Protocol         # Wire protocol encoding/decoding
```

## Next Steps

* [Getting Started](/client/getting-started) - Basic setup and usage
* [Producer API](/client/producer) - Sending messages
* [Consumer API](/client/consumer) - Receiving messages
* [ShareConsumer](/client/share-consumer) - Production-ready Share Groups (KIP-932)
* [Codecs](/client/codecs) - Type-safe serialization
* [Authentication](/client/authentication) - SASL configuration
* [Benchmarks](/client/benchmarks) - Performance comparisons

---

---
url: https://chrisrecalis.github.io/kafkats/client/getting-started.md
---
# Getting Started with @kafkats/client

## Creating a Client

The `KafkaClient` is the main entry point for all Kafka operations:

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-application',
	brokers: ['localhost:9092'],
})
```

::: tip Module formats
All `@kafkats/*` packages ship both ESM and CommonJS builds. In a CJS project, use `require` instead:

```js
const { KafkaClient } = require('@kafkats/client')
```

:::

### Client Options

| Option                      | Type         | Description                                         |
| --------------------------- | ------------ | --------------------------------------------------- |
| `clientId`                  | `string`     | Identifier for this client (appears in broker logs) |
| `brokers`                   | `string[]`   | Bootstrap broker addresses                          |
| `tls`                       | `TlsConfig`  | TLS configuration (omit for plaintext)              |
| `sasl`                      | `SaslConfig` | SASL authentication configuration                   |
| `connectionTimeoutMs`       | `number`     | Connection timeout in ms (default: 10000)           |
| `requestTimeoutMs`          | `number`     | Request timeout in ms (default: 30000)              |
| `metadataRefreshIntervalMs` | `number`     | Metadata refresh interval (default: 300000)         |
| `maxInFlightRequests`       | `number`     | Max in-flight requests per connection (default: 5)  |

### TLS Configuration

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: { enabled: true }, // Use system CA
})

// Or with custom certificates
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: {
		enabled: true,
		ca: fs.readFileSync('ca.pem'),
		cert: fs.readFileSync('client.pem'),
		key: fs.readFileSync('client-key.pem'),
	},
})
```

## Creating a Producer

```typescript
const producer = client.producer({
	acks: 'all', // Wait for all replicas
	compression: 'snappy', // Compress messages
	lingerMs: 5, // Batch for 5ms
})

// Send messages
await producer.send('my-topic', [{ value: 'Hello!' }])

// Close when done
await producer.disconnect()
```

See [Producer API](/client/producer) for full documentation.

## Creating a Consumer

```typescript
const consumer = client.consumer({
	groupId: 'my-consumer-group',
	autoOffsetReset: 'earliest',
})

// Process messages
await consumer.runEach('my-topic', async (message, ctx) => {
	console.log(`${ctx.topic}[${ctx.partition}] @ ${ctx.offset}: ${message.value}`)
})
```

See [Consumer API](/client/consumer) for full documentation.

## Creating a ShareConsumer

Kafka Share Groups (KIP-932) provide queue-like consumption with per-record acknowledgements. `ShareConsumer`
requires the production-ready Share Consumer APIs in Kafka 4.2+. Kafka 4.2.1+ is recommended because it fixes a
critical Share Group broker deadlock.

```typescript
const shareConsumer = client.shareConsumer({
	groupId: 'my-share-group',
})

await shareConsumer.runEach('my-topic', async message => {
	// For string topic subscriptions, key/value are raw Buffers (same as Consumer).
	if (message.value !== null) {
		await process(message.value.toString('utf-8'))
	}

	// If you don't call ack/release/reject, the message is implicitly ack'd (ACCEPT) on success.
	// await message.release()
	// await message.reject()
})
```

See [ShareConsumer API](/client/share-consumer) for requirements and full documentation.

## Typed Topics

Define topics with type-safe codecs:

```typescript
import { topic, string, json } from '@kafkats/client'

interface UserEvent {
	userId: string
	action: 'login' | 'logout'
	timestamp: number
}

const userEvents = topic('user-events', {
	key: string(),
	value: json<UserEvent>(),
})

// Producer - type-checked
await producer.send(userEvents, [
	{
		key: 'user-123',
		value: { userId: 'user-123', action: 'login', timestamp: Date.now() },
	},
])

// Consumer - type-inferred
await consumer.runEach(userEvents, async message => {
	// message.key is string, message.value is UserEvent
	console.log(`User ${message.value.userId} performed ${message.value.action}`)
})
```

## Error Handling

kafkats provides specific error types for different failure scenarios:

```typescript
import { KafkaError, ConnectionError, TimeoutError, isRetriable } from '@kafkats/client'

try {
	await producer.send('events', [{ value: 'data' }])
} catch (error) {
	if (error instanceof ConnectionError) {
		console.log('Connection failed:', error.message)
	} else if (error instanceof TimeoutError) {
		console.log('Request timed out')
	} else if (isRetriable(error)) {
		console.log('Retriable error, will retry automatically')
	}
}
```

## Graceful Shutdown

Always close clients when shutting down:

```typescript
process.on('SIGTERM', async () => {
	consumer.stop()
	await producer.disconnect()
	process.exit(0)
})
```

---

---
url: https://chrisrecalis.github.io/kafkats/client/configuration.md
---
# Configuration

## KafkaClient Configuration

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	// Required
	clientId: 'my-app',
	brokers: ['broker1:9092', 'broker2:9092'],

	// Optional
	tls: { enabled: true },
	sasl: { mechanism: 'PLAIN', username: 'user', password: 'pass' },
	connectionTimeoutMs: 10000,
	requestTimeoutMs: 30000,
})
```

SASL is configured based on `mechanism` (for example, `OAUTHBEARER` uses an async provider):

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['broker1:9093'],
	tls: { enabled: true },
	sasl: {
		mechanism: 'OAUTHBEARER',
		oauthBearerProvider: async () => {
			// For Amazon MSK IAM, install:
			//   pnpm add aws-msk-iam-sasl-signer-js
			const { generateAuthToken } = await import('aws-msk-iam-sasl-signer-js')
			const { token } = await generateAuthToken({ region: process.env.AWS_REGION! })
			return { value: token }
		},
	},
})
```

### Options Reference

| Option                      | Type                                     | Default  | Description                                      |
| --------------------------- | ---------------------------------------- | -------- | ------------------------------------------------ |
| `clientId`                  | `string`                                 | -        | Required. Client identifier shown in broker logs |
| `brokers`                   | `string[]`                               | -        | Required. Bootstrap broker addresses             |
| `requestTimeoutMs`          | `number`                                 | `30000`  | Request timeout (ms)                             |
| `connectionTimeoutMs`       | `number`                                 | `10000`  | Connection timeout (ms)                          |
| `metadataRefreshIntervalMs` | `number`                                 | `300000` | How often to refresh cluster metadata (ms)       |
| `maxInFlightRequests`       | `number`                                 | `5`      | Max in-flight requests per broker connection     |
| `tls`                       | `TlsConfig`                              | -        | TLS configuration (omit for plaintext)           |
| `sasl`                      | `SaslConfig`                             | -        | SASL authentication configuration                |
| `logger`                    | `Logger`                                 | -        | Custom logger implementation                     |
| `logLevel`                  | `'debug' \| 'info' \| 'warn' \| 'error'` | `'info'` | Log level for the built-in logger                |

### SASL Reauthentication

If the broker enables periodic SASL reauthentication (`connections.max.reauth.ms`), kafkats will reauthenticate automatically.
You can tune how early it refreshes via `sasl.reauthenticationThresholdMs` (default: `10000`).

## Producer Configuration

```typescript
const producer = client.producer({
	acks: 'all',
	compression: 'snappy',
	lingerMs: 5,
	maxBatchBytes: 16384,
	retries: 3,
	idempotent: false,
	transactionalId: undefined,
})
```

### Options Reference

| Option                 | Type                                              | Default     | Description                                                                                            |
| ---------------------- | ------------------------------------------------- | ----------- | ------------------------------------------------------------------------------------------------------ |
| `acks`                 | `'all' \| 'leader' \| 'none'`                     | `'all'`     | Acknowledgment mode                                                                                    |
| `compression`          | `'none' \| 'gzip' \| 'snappy' \| 'lz4' \| 'zstd'` | `'none'`    | Compression type                                                                                       |
| `lingerMs`             | `number`                                          | `5`         | Batch wait time (ms)                                                                                   |
| `maxBatchBytes`        | `number`                                          | `16384`     | Max batch size (bytes)                                                                                 |
| `retries`              | `number`                                          | `3`         | Retry attempts                                                                                         |
| `retryBackoffMs`       | `number`                                          | `100`       | Initial retry backoff (ms)                                                                             |
| `maxRetryBackoffMs`    | `number`                                          | `1000`      | Max retry backoff (ms)                                                                                 |
| `partitioner`          | `'murmur2' \| 'round-robin' \| Function`          | `'murmur2'` | Partitioning strategy                                                                                  |
| `requestTimeoutMs`     | `number`                                          | `30000`     | Request timeout (ms)                                                                                   |
| `idempotent`           | `boolean`                                         | `false`     | Enable idempotent producer                                                                             |
| `maxInFlight`          | `number`                                          | `5`         | Max in-flight requests                                                                                 |
| `transactionalId`      | `string`                                          | -           | Enable transactions                                                                                    |
| `transactionTimeoutMs` | `number`                                          | `60000`     | Transaction timeout (ms)                                                                               |
| `maxBlockMs`           | `number`                                          | `60000`     | Max time to block acquiring a producer id during transactional/idempotent producer initialization (ms) |

### Acknowledgment Modes

| Value      | Meaning                       | Durability                        |
| ---------- | ----------------------------- | --------------------------------- |
| `'none'`   | Don't wait for acknowledgment | Lowest - may lose messages        |
| `'leader'` | Wait for leader to write      | Medium - may lose if leader fails |
| `'all'`    | Wait for all in-sync replicas | Highest - recommended             |

### Compression Options

| Type       | Speed   | Ratio | Notes                    |
| ---------- | ------- | ----- | ------------------------ |
| `'none'`   | Fastest | 1:1   | No compression           |
| `'gzip'`   | Slow    | Best  | Good for text, built-in  |
| `'snappy'` | Fast    | Good  | Balanced choice          |
| `'lz4'`    | Fastest | Good  | Best for high throughput |
| `'zstd'`   | Medium  | Best  | Modern, efficient        |

GZIP is built-in. For Snappy, LZ4, and Zstd, you need to install and register a compression library. See [Compression](/client/compression) for supported libraries.

## Consumer Configuration

```typescript
const consumer = client.consumer({
	groupId: 'my-group',
	sessionTimeoutMs: 30000,
	heartbeatIntervalMs: 3000,
	autoOffsetReset: 'latest',
	isolationLevel: 'read_committed',
})
```

### Options Reference

| Option                        | Type                                          | Default                | Description                                           |
| ----------------------------- | --------------------------------------------- | ---------------------- | ----------------------------------------------------- |
| `groupId`                     | `string`                                      | -                      | Required. Consumer group ID                           |
| `groupInstanceId`             | `string`                                      | -                      | Static membership ID                                  |
| `sessionTimeoutMs`            | `number`                                      | `30000`                | Session timeout (ms)                                  |
| `rebalanceTimeoutMs`          | `number`                                      | `60000`                | Rebalance timeout (ms)                                |
| `heartbeatIntervalMs`         | `number`                                      | `3000`                 | Heartbeat interval (ms)                               |
| `maxBytesPerPartition`        | `number`                                      | `1048576`              | Max fetch bytes per partition                         |
| `maxRecords`                  | `number`                                      | `500`                  | Max records returned by one poll across partitions    |
| `minBytes`                    | `number`                                      | `1`                    | Min bytes to fetch                                    |
| `maxWaitMs`                   | `number`                                      | `5000`                 | Max fetch wait time (ms)                              |
| `autoOffsetReset`             | `'earliest' \| 'latest' \| 'none'`            | `'latest'`             | Offset reset strategy                                 |
| `isolationLevel`              | `'read_committed' \| 'read_uncommitted'`      | `'read_committed'`     | Transaction isolation                                 |
| `partitionAssignmentStrategy` | `'cooperative-sticky' \| 'sticky' \| 'range'` | `'cooperative-sticky'` | Assignment strategy                                   |
| `defaultApiTimeoutMs`         | `number`                                      | `60000`                | Max time to retry offset fetch/commit operations (ms) |

### Offset Reset Strategies

| Value        | Behavior                                    |
| ------------ | ------------------------------------------- |
| `'earliest'` | Start from beginning of topic               |
| `'latest'`   | Start from end of topic (new messages only) |
| `'none'`     | Throw error if no committed offset exists   |

### Isolation Levels

| Value                | Behavior                                  |
| -------------------- | ----------------------------------------- |
| `'read_committed'`   | Only see committed transactional messages |
| `'read_uncommitted'` | See all messages including uncommitted    |

## Environment-Based Configuration

```typescript
const client = new KafkaClient({
	clientId: process.env.KAFKA_CLIENT_ID || 'my-app',
	brokers: (process.env.KAFKA_BROKERS || 'localhost:9092').split(','),
	tls: process.env.KAFKA_TLS_ENABLED === 'true' ? { enabled: true } : undefined,
	sasl:
		process.env.KAFKA_SASL_MECHANISM === 'OAUTHBEARER'
			? {
					mechanism: 'OAUTHBEARER',
					oauthBearerProvider: async () => {
						// For Amazon MSK IAM, see the "Authentication" docs for required dependency + setup.
						const { generateAuthToken } = await import('aws-msk-iam-sasl-signer-js')
						const { token } = await generateAuthToken({ region: process.env.AWS_REGION! })
						return { value: token }
					},
				}
			: process.env.KAFKA_SASL_USERNAME
				? {
						mechanism: (process.env.KAFKA_SASL_MECHANISM || 'SCRAM-SHA-256') as 'SCRAM-SHA-256',
						username: process.env.KAFKA_SASL_USERNAME,
						password: process.env.KAFKA_SASL_PASSWORD!,
					}
				: undefined,
})
```

---

---
url: https://chrisrecalis.github.io/kafkats/client/benchmarks.md
---
# Benchmarks

The `@kafkats/benchmark` package runs throughput benchmarks comparing `@kafkats/client` against KafkaJS.

## Running

Benchmarks use a persistent Docker Compose Kafka cluster. Start it once:

```bash
# If running from a devcontainer:
KAFKA_ADVERTISED_HOST=host.docker.internal docker compose -f packages/benchmark/docker-compose.yml up -d

# If running directly on the host:
# KAFKA_ADVERTISED_HOST=localhost docker compose -f packages/benchmark/docker-compose.yml up -d
```

Then run the benchmarks:

```bash
KAFKA_BROKERS=host.docker.internal:19292,host.docker.internal:19293,host.docker.internal:19294 pnpm -C packages/benchmark bench:producer -- --iterations 10 --warmup 2
KAFKA_BROKERS=host.docker.internal:19292,host.docker.internal:19293,host.docker.internal:19294 pnpm -C packages/benchmark bench:consumer -- --iterations 10 --warmup 2
```

## Sample Results

These results were collected with:

* Kafka cluster: 3 brokers via `packages/benchmark/docker-compose.yml`
* Payload: 10,000 messages, 1 KB message size
* Producer batch size: 100
* Iterations: 10 (+2 warmup)

### Producer

| Library               | Mean throughput |
| --------------------- | --------------- |
| `@kafkats/client`     | 39,545 msg/s    |
| `kafkajs`             | 22,333 msg/s    |
| `@platformatic/kafka` | 37,636 msg/s    |

`@kafkats/client` vs KafkaJS: **1.77x** mean throughput.

### Consumer

| Library               | Mean throughput |
| --------------------- | --------------- |
| `@kafkats/client`     | 107,075 msg/s   |
| `kafkajs`             | 62,666 msg/s    |
| `@platformatic/kafka` | 73,124 msg/s    |

`@kafkats/client` vs KafkaJS: **1.71x** mean throughput.

::: tip Notes
Absolute numbers vary by hardware, Docker/VM networking, and tuning. For reproducible comparisons, run the suite on your target environment and compare ratios rather than raw msg/s.
:::

---

---
url: https://chrisrecalis.github.io/kafkats/client/migration-from-kafkajs.md
---
# Migrating from KafkaJS

This guide helps you migrate from [KafkaJS](https://kafka.js.org/) to `@kafkats/client`. While both libraries serve the same purpose, there are important API differences to be aware of.

## Why Migrate?

`@kafkats/client` offers several advantages over KafkaJS:

* **Pure TypeScript** - Written from scratch in TypeScript with full type safety
* **Type-safe topics** - Define topics with codecs for compile-time type checking
* **Modern API** - Async-first APIs like `runEach()`/`runBatch()`/`transaction()`
* **Active development** - Regular updates with modern Kafka protocol support
* **Share Groups** - Support for KIP-932 Share Groups
* **Transactions** - Full exactly-once semantics with transactional producer

## Quick Comparison

| Feature           | KafkaJS                     | @kafkats/client              |
| ----------------- | --------------------------- | ---------------------------- |
| Client creation   | `new Kafka({...})`          | `new KafkaClient({...})`     |
| Producer creation | `kafka.producer()`          | `client.producer()`          |
| Consumer creation | `kafka.consumer({groupId})` | `client.consumer({groupId})` |
| Admin creation    | `kafka.admin()`             | `client.admin()`             |
| Message handler   | `eachMessage` callback      | `runEach()` method           |
| Batch handler     | `eachBatch` callback        | `runBatch()` method          |
| Type safety       | Runtime only                | Compile-time with codecs     |

## Client Configuration

### KafkaJS

```typescript
const { Kafka, logLevel } = require('kafkajs')

const kafka = new Kafka({
	clientId: 'my-app',
	brokers: ['kafka1:9092', 'kafka2:9092'],
	connectionTimeout: 3000,
	requestTimeout: 30000,
	ssl: true,
	sasl: {
		mechanism: 'scram-sha-256',
		username: 'user',
		password: 'pass',
	},
	retry: {
		initialRetryTime: 100,
		retries: 8,
	},
	logLevel: logLevel.INFO,
})
```

### @kafkats/client

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka1:9092', 'kafka2:9092'],
	connectionTimeoutMs: 3000,
	requestTimeoutMs: 30000,
	tls: { enabled: true },
	sasl: {
		mechanism: 'SCRAM-SHA-256',
		username: 'user',
		password: 'pass',
	},
	logLevel: 'info',
})
```

### Configuration Mapping

| KafkaJS | @kafkats/client | Notes |
| ------------------- | ---------------------------------------------------------- | ------------------------------------------------------- | --------------- | ---------------- |
| `clientId` | `clientId` | Same |
| `brokers` | `brokers` | Same |
| `connectionTimeout` | `connectionTimeoutMs` | Explicit `Ms` suffix |
| `requestTimeout` | `requestTimeoutMs` | Explicit `Ms` suffix |
| `ssl: true` | `tls: { enabled: true }` | Renamed to `tls` |
| `ssl: {...}` | `tls: { enabled: true, ... }` | Same options as Node.js `tls.connect()` |
| `sasl` | `sasl` | Same structure, but mechanism is `'PLAIN'               | 'SCRAM-SHA-256' | 'SCRAM-SHA-512'` |
| `retry` | `producer({ retries, retryBackoffMs, maxRetryBackoffMs })` | Retry is configured per producer |
| `logLevel` | `logLevel` | String values: `'debug'`, `'info'`, `'warn'`, `'error'` |

## Producer Migration

### KafkaJS

```typescript
const { Partitioners, CompressionTypes } = require('kafkajs')

const producer = kafka.producer({
	createPartitioner: Partitioners.DefaultPartitioner,
	allowAutoTopicCreation: true,
	idempotent: true,
	maxInFlightRequests: 5,
})

await producer.connect()

await producer.send({
	topic: 'my-topic',
	messages: [
		{ key: 'key1', value: 'value1' },
		{ key: 'key2', value: JSON.stringify({ foo: 'bar' }), headers: { source: 'app' } },
	],
	acks: -1,
	timeout: 30000,
	compression: CompressionTypes.GZIP,
})

await producer.disconnect()
```

### @kafkats/client

```typescript
const producer = client.producer({
	partitioner: 'murmur2', // or 'round-robin' or custom function
	idempotent: true,
	maxInFlight: 5,
	acks: 'all', // 'all' | 'leader' | 'none'
	// 'gzip' works out of the box; for 'snappy'/'lz4'/'zstd' you must register a codec first
	compression: 'gzip', // 'none' | 'gzip' | 'snappy' | 'lz4' | 'zstd'
	lingerMs: 5,
	maxBatchBytes: 16384,
	retries: 3,
})

// No explicit connect() needed - connects automatically on first send

const result = await producer.send('my-topic', [
	{ key: 'key1', value: Buffer.from('value1') },
	{ key: 'key2', value: Buffer.from(JSON.stringify({ foo: 'bar' })), headers: { source: 'app' } },
])

await producer.disconnect()
```

::: tip Transactional producers
If you configure `transactionalId`, you must use `producer.transaction(...)` and cannot call `producer.send(...)` directly.
:::

### Type-Safe Producer with Codecs

```typescript
import { topic, string, json } from '@kafkats/client'

interface OrderEvent {
	orderId: string
	amount: number
	status: string
}

const orderTopic = topic('orders', {
	key: string(),
	value: json<OrderEvent>(),
})

// Type-checked at compile time
await producer.send(orderTopic, {
	key: 'order-123',
	value: { orderId: 'order-123', amount: 99.99, status: 'created' },
})
```

### Producer Configuration Mapping

| KafkaJS               | @kafkats/client    | Notes                                                                                                                         |
| --------------------- | ------------------ | ----------------------------------------------------------------------------------------------------------------------------- |
| `createPartitioner`   | `partitioner`      | `'murmur2'`, `'round-robin'`, or function                                                                                     |
| `acks: -1`            | `acks: 'all'`      | String values: `'all'`, `'leader'`, `'none'`                                                                                  |
| `acks: 1`             | `acks: 'leader'`   |                                                                                                                               |
| `acks: 0`             | `acks: 'none'`     |                                                                                                                               |
| `timeout`             | `requestTimeoutMs` | Configured on the producer (not per-send)                                                                                     |
| `compression`         | `compression`      | `gzip` is built-in; `snappy`/`lz4`/`zstd` require `npm install` + codec registration (see [Compression](/client/compression)) |
| `idempotent`          | `idempotent`       | Same                                                                                                                          |
| `transactionalId`     | `transactionalId`  | Use `producer.transaction(...)` for sends                                                                                     |
| `maxInFlightRequests` | `maxInFlight`      | Shorter name                                                                                                                  |
| N/A                   | `lingerMs`         | Batching delay (new feature)                                                                                                  |
| N/A                   | `maxBatchBytes`    | Batch size limit (new feature)                                                                                                |

### Batch Sending

**KafkaJS:**

```typescript
await producer.sendBatch({
	topicMessages: [
		{ topic: 'topic-a', messages: [{ value: 'msg1' }] },
		{ topic: 'topic-b', messages: [{ value: 'msg2' }] },
	],
})
```

**@kafkats/client:**

```typescript
// Send to multiple topics with separate calls (batched internally)
await Promise.all([
	producer.send('topic-a', [{ value: Buffer.from('msg1') }]),
	producer.send('topic-b', [{ value: Buffer.from('msg2') }]),
])
```

## Consumer Migration

### KafkaJS

```typescript
const consumer = kafka.consumer({
	groupId: 'my-group',
	sessionTimeout: 30000,
	rebalanceTimeout: 60000,
	heartbeatInterval: 3000,
	maxBytesPerPartition: 1048576,
	minBytes: 1,
	maxBytes: 10485760,
	maxWaitTimeInMs: 5000,
	readUncommitted: false,
})

await consumer.connect()
await consumer.subscribe({ topics: ['my-topic'], fromBeginning: true })

await consumer.run({
	autoCommit: true,
	autoCommitInterval: 5000,
	autoCommitThreshold: null,
	eachMessage: async ({ topic, partition, message }) => {
		console.log({
			topic,
			partition,
			offset: message.offset,
			key: message.key?.toString(),
			value: message.value?.toString(),
			headers: message.headers,
		})
	},
})
```

### @kafkats/client

```typescript
const consumer = client.consumer({
	groupId: 'my-group',
	sessionTimeoutMs: 30000,
	rebalanceTimeoutMs: 60000,
	heartbeatIntervalMs: 3000,
	maxBytesPerPartition: 1048576,
	minBytes: 1,
	maxWaitMs: 5000,
	autoOffsetReset: 'earliest', // 'earliest' | 'latest' | 'none'
	isolationLevel: 'read_committed', // 'read_committed' | 'read_uncommitted'
})

// No explicit connect() or subscribe() - all handled by runEach()

await consumer.runEach(
	'my-topic',
	async (message, ctx) => {
		console.log({
			topic: message.topic,
			partition: message.partition,
			offset: message.offset,
			key: message.key?.toString(),
			value: message.value?.toString(),
			headers: message.headers,
		})
	},
	{
		autoCommit: true,
		autoCommitIntervalMs: 5000,
	}
)
```

### Consumer Configuration Mapping

| KafkaJS                 | @kafkats/client                      | Notes                                         |
| ----------------------- | ------------------------------------ | --------------------------------------------- |
| `groupId`               | `groupId`                            | Same                                          |
| `sessionTimeout`        | `sessionTimeoutMs`                   | Explicit `Ms` suffix                          |
| `rebalanceTimeout`      | `rebalanceTimeoutMs`                 | Explicit `Ms` suffix                          |
| `heartbeatInterval`     | `heartbeatIntervalMs`                | Explicit `Ms` suffix                          |
| `maxBytesPerPartition`  | `maxBytesPerPartition`               | Same                                          |
| `minBytes`              | `minBytes`                           | Same                                          |
| `maxWaitTimeInMs`       | `maxWaitMs`                          | Shorter name                                  |
| `readUncommitted: true` | `isolationLevel: 'read_uncommitted'` | More explicit                                 |
| `fromBeginning: true`   | `autoOffsetReset: 'earliest'`        | Per-consumer config                           |
| N/A                     | `partitionAssignmentStrategy`        | `'cooperative-sticky'`, `'sticky'`, `'range'` |

### Subscribe vs Direct Topic

**KafkaJS** requires explicit subscribe:

```typescript
await consumer.subscribe({ topics: ['topic-a', 'topic-b'] })
await consumer.run({ eachMessage: handler })
```

**@kafkats/client** combines subscribe and run:

```typescript
// Single topic
await consumer.runEach('my-topic', handler)

// Multiple topics
await consumer.runEach(['topic-a', 'topic-b'], handler)
```

### Batch Processing

**KafkaJS:**

```typescript
await consumer.run({
	eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning }) => {
		for (const message of batch.messages) {
			await processMessage(message)
			resolveOffset(message.offset)
			await heartbeat()
		}
	},
})
```

**@kafkats/client:**

```typescript
await consumer.runBatch(
	'my-topic',
	async (messages, ctx) => {
		for (const message of messages) {
			await processMessage(message)
		}
		// Offsets are committed after batch completes successfully
	},
	{
		autoCommit: true,
		autoCommitIntervalMs: 5000,
	}
)
```

### Type-Safe Consumer

```typescript
import { topic, string, json } from '@kafkats/client'

interface OrderEvent {
	orderId: string
	amount: number
}

const orderTopic = topic('orders', {
	key: string(),
	value: json<OrderEvent>(),
})

// message.key is string, message.value is OrderEvent
await consumer.runEach(orderTopic, async message => {
	console.log(`Order ${message.value.orderId}: $${message.value.amount}`)
})
```

### Consumer Events

**KafkaJS:**

```typescript
const { GROUP_JOIN, REBALANCING } = consumer.events

consumer.on(GROUP_JOIN, e => {
	console.log('Joined group', e.payload.groupId)
})

consumer.on(REBALANCING, e => {
	console.log('Rebalancing...', e.payload.groupId)
})
```

**@kafkats/client:**

```typescript
consumer.on('running', () => {
	console.log('Consumer started')
})

consumer.on('partitionsAssigned', partitions => {
	console.log('Assigned partitions:', partitions)
})

consumer.on('partitionsRevoked', partitions => {
	console.log('Revoked partitions:', partitions)
})

consumer.on('stopped', () => {
	console.log('Consumer stopped')
})
```

### Pause/Resume

**KafkaJS:**

```typescript
consumer.pause([{ topic: 'my-topic', partitions: [0, 1] }])
consumer.resume([{ topic: 'my-topic', partitions: [0, 1] }])
```

**@kafkats/client:**

```typescript
consumer.pause([
	{ topic: 'my-topic', partition: 0 },
	{ topic: 'my-topic', partition: 1 },
])
consumer.resume([
	{ topic: 'my-topic', partition: 0 },
	{ topic: 'my-topic', partition: 1 },
])
```

### Stopping the Consumer

**KafkaJS:**

```typescript
await consumer.disconnect()
```

**@kafkats/client:**

```typescript
consumer.stop() // Signals graceful shutdown
// The runEach/runBatch promise resolves when stopped
```

## Transactions

### KafkaJS

```typescript
const producer = kafka.producer({
	transactionalId: 'my-txn-producer',
	maxInFlightRequests: 1,
	idempotent: true,
})

await producer.connect()

const transaction = await producer.transaction()

try {
	await transaction.send({ topic: 'topic-a', messages: [{ value: 'msg' }] })

	await transaction.sendOffsets({
		consumerGroupId: 'my-group',
		topics: [{ topic: 'input-topic', partitions: [{ partition: 0, offset: '100' }] }],
	})

	await transaction.commit()
} catch (e) {
	await transaction.abort()
	throw e
}
```

### @kafkats/client

```typescript
const producer = client.producer({
	transactionalId: 'my-txn-producer',
	maxInFlight: 1,
	idempotent: true,
})

// Automatic commit/abort based on callback success/failure
await producer.transaction(async tx => {
	await tx.send('topic-a', [{ value: Buffer.from('msg') }])

	await tx.sendOffsets({
		groupId: 'my-group',
		offsets: [{ topic: 'input-topic', partition: 0, offset: 100n }],
	})
})
```

## Admin API

### KafkaJS

```typescript
const admin = kafka.admin()
await admin.connect()

// List topics
const topics = await admin.listTopics()

// Create topics
await admin.createTopics({
	topics: [{ topic: 'new-topic', numPartitions: 3, replicationFactor: 2 }],
})

// Delete topics
await admin.deleteTopics({ topics: ['old-topic'] })

// Describe groups
const groups = await admin.describeGroups(['my-group'])

// Delete groups
await admin.deleteGroups(['old-group'])

await admin.disconnect()
```

### @kafkats/client

```typescript
const admin = client.admin()

// List topics
const topics = await admin.listTopics()

// Create topics
await admin.createTopics([{ name: 'new-topic', numPartitions: 3, replicationFactor: 2 }])

// Delete topics
await admin.deleteTopics(['old-topic'])

// Describe groups
const groups = await admin.describeGroups(['my-group'])

// Delete groups
await admin.deleteGroups(['old-group'])

// No `admin.disconnect()`; call `client.disconnect()` when shutting down
```

### Admin API Mapping

| KafkaJS                        | @kafkats/client              | Notes                  |
| ------------------------------ | ---------------------------- | ---------------------- |
| `admin.connect()`              | N/A                          | No explicit connect    |
| `admin.disconnect()`           | N/A                          | No explicit disconnect |
| `admin.listTopics()`           | `admin.listTopics()`         | Same                   |
| `admin.createTopics({topics})` | `admin.createTopics(topics)` | Direct array           |
| `admin.deleteTopics({topics})` | `admin.deleteTopics(topics)` | Direct array           |
| `admin.fetchTopicMetadata()`   | `admin.describeTopics()`     | Renamed                |
| `admin.listGroups()`           | `admin.listGroups()`         | Same                   |
| `admin.describeGroups()`       | `admin.describeGroups()`     | Same                   |
| `admin.deleteGroups()`         | `admin.deleteGroups()`       | Same                   |
| `admin.describeCluster()`      | `admin.describeCluster()`    | Same                   |
| `admin.fetchTopicOffsets()`    | `admin.fetchTopicOffsets()`  | Similar                |
| `admin.describeAcls()`         | `admin.describeAcls()`       | Same                   |
| `admin.createAcls()`           | `admin.createAcls()`         | Same                   |
| `admin.deleteAcls()`           | `admin.deleteAcls()`         | Same                   |

## Compression

### KafkaJS

```typescript
const { CompressionTypes, CompressionCodecs } = require('kafkajs')
const SnappyCodec = require('kafkajs-snappy')

CompressionCodecs[CompressionTypes.Snappy] = SnappyCodec

await producer.send({
	topic: 'my-topic',
	compression: CompressionTypes.GZIP,
	messages: [{ value: 'data' }],
})
```

### @kafkats/client

```typescript
// GZIP works out of the box. For Snappy/LZ4/Zstd, just install a supported
// library (e.g. `npm install snappy`) — it is detected and registered
// automatically, no registration code needed.
const producer = client.producer({
	compression: 'snappy', // 'none' | 'gzip' | 'snappy' | 'lz4' | 'zstd'
})

await producer.send('my-topic', [{ value: Buffer.from('data') }])
```

See the [Compression docs](/client/compression) for more details and supported libraries.

## Error Handling

### KafkaJS

```typescript
const { KafkaJSError, KafkaJSConnectionError } = require('kafkajs')

try {
	await producer.send({ topic: 'my-topic', messages: [] })
} catch (error) {
	if (error instanceof KafkaJSConnectionError) {
		console.log('Connection error')
	} else if (error.retriable) {
		console.log('Retriable error')
	}
}
```

### @kafkats/client

```typescript
import { KafkaError, KafkaProtocolError, ConnectionError, TimeoutError, isRetriable } from '@kafkats/client'

try {
	await producer.send('my-topic', [{ value: Buffer.from('data') }])
} catch (error) {
	if (error instanceof ConnectionError) {
		console.log('Connection error')
	} else if (error instanceof TimeoutError) {
		console.log('Request timed out')
	} else if (error instanceof KafkaProtocolError) {
		console.log('Protocol error:', error.errorCode)
	} else if (isRetriable(error)) {
		console.log('Retriable error')
	}
}
```

## Graceful Shutdown

### KafkaJS

```typescript
const shutdown = async () => {
	await consumer.disconnect()
	await producer.disconnect()
	await admin.disconnect()
}

process.on('SIGTERM', shutdown)
process.on('SIGINT', shutdown)
```

### @kafkats/client

```typescript
const consumerRun = consumer.runEach('my-topic', async message => {
	// ...
})

const shutdown = async () => {
	consumer.stop() // Graceful stop
	await consumerRun.catch(() => {}) // Optional: wait for runEach/runBatch to exit
	await producer.disconnect()
	await client.disconnect() // Close all broker connections
}

process.on('SIGTERM', shutdown)
process.on('SIGINT', shutdown)
```

## Migration Checklist

1. **Update imports**
   * `const { Kafka } = require('kafkajs')` → `import { KafkaClient } from '@kafkats/client'`

2. **Update client configuration**
   * `new Kafka({...})` → `new KafkaClient({...})`
   * Add `Ms` suffix to timeout options
   * Change `ssl` to `tls`

3. **Update producer code**
   * Remove `await producer.connect()`
   * Change message values to `Buffer` (or use typed topics)
   * Update `acks` to string values
   * Update compression to string values (`'gzip'` works out of the box; for `'snappy'`/`'lz4'`/`'zstd'` you must `npm install` a library and register the codec)

4. **Update consumer code**
   * Replace `subscribe()` + `run()` with `runEach()` or `runBatch()`
   * Change `fromBeginning` to `autoOffsetReset: 'earliest'`
   * Update event listeners

5. **Update admin code**
   * Remove `connect()` and `disconnect()` calls
   * Simplify method arguments (direct arrays instead of objects)

6. **Update error handling**
   * Import new error types
   * Update `instanceof` checks

7. **Update compression setup**
   * GZIP works without any packages
   * For Snappy/LZ4/Zstd: install a compression library (e.g. `npm install snappy`) — it is registered automatically. See [Compression docs](/client/compression)

## Feature Differences

### Features in @kafkats/client not in KafkaJS

* **Type-safe topics** with compile-time checking
* **Share Groups** (KIP-932) for queue-like consumption
* **Cooperative sticky assignor** by default
* **Stream mode** with async iterators

### Features in KafkaJS not yet in @kafkats/client

* `consumer.commitOffsets()` for manual offset commits outside handlers
* Admin partition reassignment (`alterPartitionReassignments`)
* Advanced retry options (e.g., `multiplier`, `factor` for exponential backoff)

## Getting Help

If you encounter issues during migration:

* Check the [full documentation](/client/)
* Review the [examples](/examples/)
* Open an issue on [GitHub](https://github.com/chrisrecalis/kafkats/issues)

---

---
url: https://chrisrecalis.github.io/kafkats/client/producer.md
---
# Producer API

The producer handles sending messages to Kafka topics with automatic batching, compression, and retries.

Unlike a “send one request per call” API, `producer.send()` is **queue-based**: messages are first queued in an in-memory accumulator, then flushed to Kafka as partition batches based on your batching settings.

## Creating a Producer

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})

const producer = client.producer({
	acks: 'all', // Wait for all replicas
	compression: 'snappy', // Compress batches
	lingerMs: 5, // Batch for 5ms
})
```

## Sending Messages

### Basic Usage

```typescript
// Send a single message
await producer.send('my-topic', [{ value: 'Hello, Kafka!' }])

// Send multiple messages
await producer.send('events', [
	{ key: 'user-1', value: 'event-1' },
	{ key: 'user-2', value: 'event-2' },
])

// Send to a different topic with a separate call
await producer.send('logs', [{ value: 'log message' }])
```

### How `send()` Works (Queue + Batches)

When you call `producer.send(...)`:

1. Messages are **encoded** (codecs / strings / buffers), assigned a **partition**, and appended to an in-memory **per-topic-partition batch**.
2. Batches are flushed to the broker when they become “ready” (see triggers below).
3. The returned promise resolves when the broker acknowledges the produced records (based on `acks`).

This design makes `send()` fast under load, but it also means you should think about **backpressure**: if you produce faster than Kafka can accept, the in-memory queue can grow.

#### Flush Triggers

| Trigger             | Controlled by                                | What happens                                                    |
| ------------------- | -------------------------------------------- | --------------------------------------------------------------- |
| Time-based batching | `lingerMs`                                   | Flush the current batch for a partition after the timer expires |
| Size-based batching | `maxBatchBytes`                              | Flush the current batch when it reaches the size threshold      |
| Explicit flush      | `producer.flush()` / `producer.disconnect()` | Flush all batches immediately and wait for acknowledgments      |

::: tip Fire-and-forget
You can call `producer.send(...)` without awaiting it, but you must eventually call `await producer.flush()` or `await producer.disconnect()` (or keep the process alive) to ensure queued records are actually delivered.
:::

### Message Structure

```typescript
interface ProducerMessage<V = Buffer, K = Buffer | string> {
	key?: K | null // Message key (determines partition)
	value: V // Message value
	headers?: Record<string, string | Buffer> // Optional headers
	partition?: number // Explicit partition (bypasses partitioner)
	timestamp?: Date // Message timestamp (defaults to now)
}
```

### Send Result

```typescript
const result = await producer.send('my-topic', { value: 'Hello!' })

console.log({
	topic: result.topic, // 'my-topic'
	partition: result.partition, // 0
	offset: result.offset, // 42n (bigint)
	timestamp: result.timestamp, // Date
})
```

## Typed Topics

Use the `topic()` helper for type-safe producers:

```typescript
import { topic, string, json } from '@kafkats/client'

interface Order {
	id: string
	items: string[]
	total: number
}

const ordersTopic = topic('orders', {
	key: string(),
	value: json<Order>(),
})

// Type-checked at compile time
await producer.send(ordersTopic, [
	{
		key: 'order-123',
		value: { id: 'order-123', items: ['item-a'], total: 99.99 },
	},
])
```

## Message Keys and Partitioning

Keys determine which partition a message goes to. Messages with the same key always go to the same partition:

```typescript
// All messages for user-1 go to the same partition (ordered)
await producer.send('events', [
	{ key: 'user-1', value: 'login' },
	{ key: 'user-1', value: 'click' },
	{ key: 'user-1', value: 'logout' },
])
```

### Partitioner Strategies

```typescript
// Default: murmur2 (consistent hashing)
const producer = client.producer({
	partitioner: 'murmur2',
})

// Round-robin (even distribution)
const producer = client.producer({
	partitioner: 'round-robin',
})

// Custom partitioner
const producer = client.producer({
	partitioner: (topic, key, value, partitionCount) => {
		if (key === null) return -1 // Use sticky partitioner
		const hash = customHash(key)
		return Math.abs(hash) % partitionCount
	},
})
```

## Batching

The producer batches messages for efficiency. Configure batching behavior:

```typescript
const producer = client.producer({
	lingerMs: 5, // Wait up to 5ms to batch messages
	maxBatchBytes: 16384, // Flush when batch reaches 16KB
})
```

### Flushing

Force all pending messages to be sent:

```typescript
// Flush and wait for all acknowledgments
await producer.flush()
```

## Compression

Enable compression to reduce network bandwidth:

```typescript
const producer = client.producer({
	compression: 'snappy', // Also: 'gzip', 'lz4', 'zstd', 'none'
})
```

| Type       | Speed     | Compression Ratio |
| ---------- | --------- | ----------------- |
| `'none'`   | Fastest   | 1:1               |
| `'snappy'` | Fast      | Good              |
| `'lz4'`    | Very fast | Good              |
| `'gzip'`   | Slow      | Best              |
| `'zstd'`   | Medium    | Best              |

::: tip Compression Libraries
GZIP is built-in. For Snappy, LZ4, and Zstd, you need to install and register a compression library. See [Compression](/client/compression) for supported libraries and setup instructions.
:::

## Error Handling and Retries

The producer automatically retries on retriable errors:

```typescript
const producer = client.producer({
	retries: 3, // Retry up to 3 times
	retryBackoffMs: 100, // Start with 100ms backoff
	maxRetryBackoffMs: 1000, // Max 1s backoff
})
```

Handle errors:

```typescript
import { SendTimeoutError, RecordTooLargeError } from '@kafkats/client'

try {
	await producer.send('my-topic', { value: largePayload })
} catch (error) {
	if (error instanceof RecordTooLargeError) {
		console.log('Message too large for broker')
	} else if (error instanceof SendTimeoutError) {
		console.log('Send timed out')
	}
}
```

## Idempotent Producer

Enable exactly-once delivery semantics:

```typescript
const producer = client.producer({
	idempotent: true,
	acks: 'all', // Required
})
```

With idempotent mode:

* Broker assigns a unique producer ID
* Per-partition sequence numbers detect duplicates
* Retries are safe and won't create duplicates

## Headers

Attach metadata to messages:

```typescript
await producer.send('events', [
	{
		value: 'event data',
		headers: {
			'correlation-id': '12345',
			source: 'web-app',
			timestamp: Date.now().toString(),
		},
	},
])
```

## Closing the Producer

Always close the producer when done:

```typescript
// Flush pending messages and close
await producer.disconnect()
```

## Producer Options

This is a quick reference for `client.producer({...})`. For the complete configuration (including consumer + client options), see [Configuration](/client/configuration).

| Option                 | Type                                                                          | Default     | Notes                                                  |
| ---------------------- | ----------------------------------------------------------------------------- | ----------- | ------------------------------------------------------ |
| `acks`                 | `'all' \| 'leader' \| 'none'`                                                 | `'all'`     | Durability vs latency tradeoff                         |
| `compression`          | `'none' \| 'gzip' \| 'snappy' \| 'lz4' \| 'zstd'`                             | `'none'`    | Applied to record batches                              |
| `lingerMs`             | `number`                                                                      | `5`         | Time-based batching                                    |
| `maxBatchBytes`        | `number`                                                                      | `16384`     | Size-based batching                                    |
| `retries`              | `number`                                                                      | `3`         | Retries on retriable errors                            |
| `retryBackoffMs`       | `number`                                                                      | `100`       | Backoff start                                          |
| `maxRetryBackoffMs`    | `number`                                                                      | `1000`      | Backoff cap                                            |
| `partitioner`          | `'murmur2' \| 'round-robin' \| (topic, key, value, partitionCount) => number` | `'murmur2'` | `-1` means “sticky” for keyless records                |
| `requestTimeoutMs`     | `number`                                                                      | `30000`     | Produce request timeout                                |
| `maxInFlight`          | `number`                                                                      | `5`         | Limits concurrent in-flight produce requests           |
| `idempotent`           | `boolean`                                                                     | `false`     | Safe retries + duplicate detection                     |
| `transactionalId`      | `string`                                                                      | -           | Enables transactions (use `producer.transaction(...)`) |
| `transactionTimeoutMs` | `number`                                                                      | `60000`     | Transaction timeout (broker + client)                  |

## Next Steps

* [Transactions](/client/transactions) - Exactly-once semantics
* [Codecs](/client/codecs) - Custom serialization
* [Error Handling](/client/errors) - Error types and handling

---

---
url: https://chrisrecalis.github.io/kafkats/client/transactions.md
---
# Transactions

kafkats supports Kafka transactions for exactly-once semantics (EOS).

## Overview

Transactions ensure that a group of messages are either all committed or all rolled back. This is essential for:

* **Exactly-once processing** - No duplicates or lost messages
* **Atomic multi-topic writes** - All-or-nothing across topics
* **Consume-transform-produce** - Atomic read-process-write patterns

## Enabling Transactions

Create a transactional producer:

```typescript
const producer = client.producer({
	transactionalId: 'my-transaction-id',
	acks: 'all', // Required for transactions
})
```

::: tip Transactional ID
The `transactionalId` must be unique per producer instance. Use a stable identifier like `${applicationName}-${instanceId}`.
:::

## Basic Transaction

```typescript
await producer.transaction(async txn => {
	// All sends in this callback are part of the transaction
	await txn.send('orders', [{ key: 'order-1', value: JSON.stringify({ status: 'created' }) }])

	await txn.send('inventory', [{ key: 'item-1', value: JSON.stringify({ delta: -1 }) }])

	// Transaction commits automatically when callback completes
})
```

If an error is thrown, the transaction is automatically aborted:

```typescript
await producer.transaction(async txn => {
	await txn.send('orders', [{ value: 'order-data' }])

	if (someCondition) {
		throw new Error('Abort transaction')
		// Transaction is rolled back, no messages are committed
	}
})
```

## Transaction Timeout

Configure transaction timeout:

```typescript
const producer = client.producer({
	transactionalId: 'my-txn',
	transactionTimeoutMs: 60000, // 60 seconds (default)
})
```

Use the abort signal for long-running operations:

```typescript
await producer.transaction(async txn => {
	// Cancel fetch if transaction times out
	const data = await fetch(url, { signal: txn.signal })
	await txn.send('results', [{ value: data }])
})
```

## Consume-Transform-Produce

For exactly-once stream processing, commit consumer offsets within the transaction:

```typescript
const consumer = client.consumer({
	groupId: 'my-group',
	isolationLevel: 'read_committed', // Only read committed messages
})

const producer = client.producer({
	transactionalId: 'my-processor',
	acks: 'all',
})

await consumer.runEach(
	'input',
	async (message, ctx) => {
		// Process message
		const result = await transform(message.value)

		// Atomically: send output + commit input offset
		await producer.transaction(async txn => {
			await txn.send('output', [{ value: result }])
			await txn.sendOffsets(ctx)
		})
	},
	{ autoCommit: false, commitOffsets: false }
)
```

`sendOffsets(ctx)` commits `ctx.offset + 1` for the context's topic and partition. The context contains a
delivery-time snapshot of the consumer-group membership that delivered the record, so a rebalance makes a stale
transaction fail instead of committing offsets under a newer generation.

::: tip Batched processing
Pass explicit accumulated offsets as the second argument: `txn.sendOffsets(ctx, offsets)`. Use the first context
that opened the transaction so the entire batch remains bound to that consumer-group generation.

`@kafkats/flow` with `processingGuarantee: 'exactly_once'` handles this batching automatically. See
[Flow Processing Guarantees](/flow/getting-started#processing-guarantees) for details.
:::

## Concurrent Transactions

A producer can have one open transaction at a time — this is a Kafka protocol constraint, not a library limit. Concurrent `transaction()` calls **queue and wait for capacity** instead of throwing, so a single transactional producer is safe to share across handlers running with `partitionConcurrency` greater than 1:

```typescript
await consumer.runBatch(
	'input',
	async (messages, ctx) => {
		const results = await transform(messages)

		// Safe with partitionConcurrency > 1: transactions from concurrent
		// partition handlers wait their turn on the producer.
		await producer.transaction(async txn => {
			await txn.send('output', results)
			await txn.sendOffsets({
				consumerGroupMetadata, // group id + generation id + member id (KIP-447 zombie fencing)
				offsets: [{ topic: ctx.topic, partition: ctx.partition, offset: ctx.offset + 1n }],
			})
		})
	},
	// commitOffsets: false — offsets must only be committed through the
	// transaction, never by the consumer itself (e.g. during revoke/shutdown).
	{ autoCommit: false, commitOffsets: false, partitionConcurrency: 3 }
)
```

For strict exactly-once semantics, pass `consumerGroupMetadata` rather than a bare `groupId` — without the generation and member id, a zombie consumer's offset commits are not fenced (see the tip in [Consume-Transform-Produce](#consume-transform-produce)).

Things to know:

* **Only the transaction section serializes.** Message processing before `transaction()` still overlaps across partitions.
* **No ordering guarantee** between independent transactions. Per-partition ordering is preserved because `runEach`/`runBatch` don't deliver the next record for a partition until the handler returns.
* **The transaction timeout does not include queue time.** It starts when the transaction actually begins.
* **Nested transactions throw.** Calling `producer.transaction()` from inside the same producer's transaction callback throws immediately instead of deadlocking.
* **Watch for commit-bound throughput.** The producer emits `transaction:queued` (with the number of transactions ahead) whenever a call has to wait. If this fires sustainedly, your throughput ceiling is transaction commit latency — increase batch sizes to lower the transaction rate, or raise `transactionConcurrency` (below) so transactions actually overlap.

```typescript
producer.on('transaction:queued', ({ queued }) => {
	metrics.gauge('kafka.txn.queue_depth', queued)
})
```

## Parallel Transactions

The one-open-transaction limit is per transactional ID. `transactionConcurrency` lets a single producer run up to N transactions at once by managing a pool of N internal transactional producers ("lanes") behind the same `transaction()` API:

```typescript
const producer = client.producer({
	transactionalId: 'orders-processor',
	transactionConcurrency: 3, // up to 3 transactions in flight
})

await consumer.runBatch(
	'input',
	async (messages, ctx) => {
		const results = await transform(messages)

		// With transactionConcurrency: 3, transactions from the 3 concurrent
		// partition handlers genuinely overlap instead of queueing.
		await producer.transaction(async txn => {
			await txn.send('output', results)
			await txn.sendOffsets({
				consumerGroupMetadata,
				offsets: [{ topic: ctx.topic, partition: ctx.partition, offset: ctx.offset + 1n }],
			})
		})
	},
	{ autoCommit: false, commitOffsets: false, partitionConcurrency: 3 }
)
```

How it works:

* Lane 0 uses `transactionalId` verbatim; lanes 1..N-1 append `-1`..`-{N-1}` (e.g. `orders-processor`, `orders-processor-1`, `orders-processor-2`). Each lane initializes lazily on first use and fences its predecessor with the same ID.
* `transaction()` calls are admitted **first-in first-out**: a call takes a free lane immediately, or waits for the next one released. `transaction:queued` fires only when all lanes are busy.
* Transactions begin in call order but **may commit in any order** — same guarantee as independent transactions today. Per-partition input ordering is unaffected (`runEach`/`runBatch` still deliver one record/batch per partition at a time).
* `flush()` and `disconnect()` cover all lanes; disconnect still refuses while any transaction is active or queued.

::: warning Fencing and transactional IDs
Transactional-ID zombie fencing applies **per lane**. For consume-transform-produce, always pass `consumerGroupMetadata` to `sendOffsets()` (KIP-447) so a zombie's offset commits are fenced by consumer-group generation regardless of which lane they ran on. Also note that changing `transactionConcurrency` changes the set of transactional IDs in use: a transaction left open under an ID that is no longer used stays open until the broker's `transactional.id.expiration.ms` elapses.
:::

## Transaction API

### ProducerTransaction

```typescript
interface ProducerTransaction {
	// Send messages within the transaction
	send(topic: string, messages: ProducerMessage[]): Promise<SendResult[]>
	send<V, K>(topicDef: TopicDefinition<V, K>, messages: ProducerMessage<V, K>[]): Promise<SendResult[]>

	// Commit consumer offsets (for exactly-once)
	sendOffsets(ctx: ConsumeContext): Promise<void>
	sendOffsets(ctx: ConsumeContext, offsets: TopicPartitionOffset[]): Promise<void>
	sendOffsets(params: SendOffsetsParams): Promise<void>

	// Abort signal (fires on timeout or error)
	readonly signal: AbortSignal
}
```

### SendOffsetsParams

```typescript
interface SendOffsetsParams {
	// Low-level, unfenced form
	groupId?: string

	// Advanced escape hatch for callers managing raw group metadata
	consumerGroupMetadata?: {
		groupId: string
		generationId: number
		memberId: string
		groupInstanceId?: string
	}

	// Offsets to commit
	offsets: Array<{
		topic: string
		partition: number
		offset: bigint
	}>
}
```

A context from a manually assigned consumer automatically uses the `groupId`-only form because there is no group
generation to fence. The object form remains available for standalone producer transactions and other low-level
callers without a consume context. For normal `runEach()`, `runBatch()`, and `stream()` processing, pass the
`ConsumeContext` supplied to the handler.

## Idempotent Producer

For simpler exactly-once delivery (without full transactions):

```typescript
const producer = client.producer({
	idempotent: true,
	acks: 'all',
})

// Retries are safe - no duplicates
await producer.send('events', { value: 'data' })
```

Idempotent mode:

* Assigns a unique producer ID
* Uses sequence numbers per partition
* Safe retries without duplicates
* Does NOT support atomic multi-topic writes

## Error Handling

### Transaction Aborted

```typescript
import { InvalidTxnStateError } from '@kafkats/client'

try {
	await producer.transaction(async txn => {
		await txn.send('topic', [{ value: 'data' }])
		// ... long operation
	})
} catch (error) {
	if (error instanceof InvalidTxnStateError) {
		// Transaction was aborted (timeout, fenced, etc.)
	}
}
```

### Producer Fenced

When another producer with the same `transactionalId` starts:

```typescript
import { ProducerFencedError } from '@kafkats/client'

try {
	await producer.transaction(async txn => {
		// ...
	})
} catch (error) {
	if (error instanceof ProducerFencedError) {
		// Another producer took over - shut down this instance
		await producer.disconnect()
		process.exit(1)
	}
}
```

## Best Practices

1. **Use stable transactional IDs** - Based on application + instance identity
2. **Keep transactions short** - Avoid long-running operations inside
3. **Use abort signal** - Cancel external operations on transaction abort
4. **Handle fencing** - Shut down gracefully when fenced
5. **Read committed** - Use `isolationLevel: 'read_committed'` for consumers

## Consumer Isolation Levels

Configure how consumers see transactional messages:

```typescript
// Only committed transactions (default, recommended)
const consumer = client.consumer({
	groupId: 'my-group',
	isolationLevel: 'read_committed',
})

// All messages including uncommitted
const consumer = client.consumer({
	groupId: 'my-group',
	isolationLevel: 'read_uncommitted',
})
```

---

---
url: https://chrisrecalis.github.io/kafkats/client/consumer.md
---
# Consumer API

The consumer reads records from Kafka topics using consumer groups for automatic partition assignment and offset commits.

## Creating a Consumer

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})

const consumer = client.consumer({
	groupId: 'my-consumer-group',
	autoOffsetReset: 'earliest',
})
```

## Subscriptions

`runEach()`, `runBatch()`, and `stream()` take a `subscription` argument.

| Subscription               | What it does                       | Example                                                   |
| -------------------------- | ---------------------------------- | --------------------------------------------------------- |
| Topic name                 | Consume raw `Buffer` key/value     | `consumer.runEach('events', handler)`                     |
| Multiple topics            | Consume multiple topics at once    | `consumer.runEach(['events', 'logs'], handler)`           |
| Typed topic (`topic(...)`) | Decode with codecs and infer types | `consumer.runEach(userEvents, handler)`                   |
| Custom subscription        | Provide explicit decoders          | `consumer.runEach({ topic: 'events', decoder }, handler)` |

## Processing Messages

### Single Message Mode (runEach)

Process messages one at a time:

```typescript
await consumer.runEach('events', async (message, ctx) => {
	console.log({
		topic: ctx.topic,
		partition: ctx.partition,
		offset: ctx.offset,
		key: message.key?.toString(),
		value: message.value?.toString(),
	})
})
```

### Batch Mode (runBatch)

Process messages in batches for higher throughput:

```typescript
await consumer.runBatch(
	'events',
	async (messages, ctx) => {
		console.log(`Received ${messages.length} messages from ${ctx.topic}[${ctx.partition}]`)

		for (const message of messages) {
			await processMessage(message)
		}
	},
	{
		maxBatchSize: 100, // Max messages per batch
		maxBatchWaitMs: 50, // Max wait time to fill batch
	}
)
```

### Async Iterator Mode (stream)

Consume messages via `for await ... of`:

```typescript
for await (const { message, ctx } of consumer.stream('events')) {
	console.log(ctx.topic, ctx.partition, ctx.offset, message.value)
}
```

## Message Structure

```typescript
interface Message<V = Buffer, K = Buffer> {
	topic: string // Source topic
	partition: number // Source partition
	offset: bigint // Message offset
	timestamp: bigint // Message timestamp (ms)
	key: K | null // Message key
	value: V // Message value
	headers: Record<string, Buffer> // Message headers
}
```

## Consume Context

```typescript
interface ConsumeContext {
	signal: AbortSignal // Aborted when consumer shuts down
	topic: string // Current message topic
	partition: number // Current message partition
	offset: bigint // Current message offset
	groupId: string // Consumer group whose offsets are being tracked
	consumerGroupMetadata?: Readonly<{
		groupId: string
		generationId: number
		memberId: string
		groupInstanceId?: string | null
	}> // Present for consumer-group delivery; captured when the record or batch is delivered
}
```

Pass the context to `transaction.sendOffsets(ctx)` when atomically committing consumed offsets with produced
records. Its consumer-group metadata is a delivery-time snapshot, so a transaction committed after a rebalance is
fenced using the membership that actually delivered the records. The fields are public and survive object spread.

Manually assigned consumers have a `groupId` but no `consumerGroupMetadata`, because they do not join the group.

Use the signal to cancel long-running operations:

```typescript
await consumer.runEach('events', async (message, ctx) => {
	const response = await fetch(url, { signal: ctx.signal })
	// ...
})
```

## Typed Consumers

Use typed topics for automatic deserialization:

```typescript
import { topic, string, json } from '@kafkats/client'

interface UserEvent {
	userId: string
	action: string
}

const userEvents = topic('user-events', {
	key: string(),
	value: json<UserEvent>(),
})

await consumer.runEach(userEvents, async message => {
	// message.key: string
	// message.value: UserEvent
	console.log(`User ${message.value.userId}: ${message.value.action}`)
})
```

## Run Options

### runEach Options

| Option                 | Type                 | Default | Description                                         |
| ---------------------- | -------------------- | ------- | --------------------------------------------------- |
| `partitionConcurrency` | `number`             | `1`     | How many partitions process concurrently            |
| `autoCommit`           | `boolean`            | `true`  | Enable periodic commits                             |
| `commitOffsets`        | `boolean`            | `true`  | Track consumed offsets for committing               |
| `autoCommitIntervalMs` | `number`             | `5000`  | Commit interval when `autoCommit` is enabled        |
| `signal`               | `AbortSignal`        | -       | Abort to stop the consumer                          |
| `assignment`           | `ManualAssignment[]` | -       | Manually assign partitions instead of joining group |

### runBatch Options

| Option                 | Type                 | Default | Description                                         |
| ---------------------- | -------------------- | ------- | --------------------------------------------------- |
| `partitionConcurrency` | `number`             | `1`     | How many partitions process concurrently            |
| `autoCommit`           | `boolean`            | `true`  | Enable periodic commits                             |
| `commitOffsets`        | `boolean`            | `true`  | Track consumed offsets for committing               |
| `autoCommitIntervalMs` | `number`             | `5000`  | Commit interval when `autoCommit` is enabled        |
| `signal`               | `AbortSignal`        | -       | Abort to stop the consumer                          |
| `maxBatchSize`         | `number`             | `100`   | Maximum messages per partition-batch                |
| `maxBatchWaitMs`       | `number`             | `50`    | Max time to wait before flushing a batch            |
| `assignment`           | `ManualAssignment[]` | -       | Manually assign partitions instead of joining group |

## Partition Concurrency

Control how many partitions are processed concurrently:

```typescript
// Process up to 4 partitions at the same time
await consumer.runEach('events', handler, {
	partitionConcurrency: 4,
})
```

::: warning
Higher concurrency increases throughput but may cause out-of-order processing across partitions. Within a partition, order is always preserved.
:::

## Offset Management

When `autoCommit: true` and `commitOffsets: true`, the consumer:

* Marks offsets as “consumed” after your handler completes successfully.
* Commits pending offsets periodically (`autoCommitIntervalMs`) and once more during shutdown (unless the session is lost).

If you set `autoCommit: false` (or `commitOffsets: false`), the consumer will not commit offsets. On restart it will resume from the last committed offsets (or apply `autoOffsetReset` if none exist).

::: tip Manual commits
The public consumer API currently focuses on automatic commits. If you need explicit offset commits, consider managing offsets externally until manual commit APIs are exposed.
:::

## Backpressure: Pause and Resume

Pause fetching from specific partitions while you're overloaded:

```typescript
consumer.pause([{ topic: 'events', partition: 0 }])
// ... catch up ...
consumer.resume([{ topic: 'events', partition: 0 }])
```

## Seeking to a Specific Offset

Reposition the consumer to a specific offset for a partition:

```typescript
// Seek to a specific offset
consumer.seek('events', 0, 100n)

// Seek to the beginning
consumer.seek('events', 0, 0n)
```

The seek only affects the **next fetch** - messages already fetched will still be delivered. For controlled seeking, combine with pause/resume:

```typescript
await consumer.runEach('events', async (message, ctx) => {
	if (shouldReplay(message)) {
		// Pause, seek, resume pattern for controlled repositioning
		consumer.pause([{ topic: ctx.topic, partition: ctx.partition }])
		consumer.seek(ctx.topic, ctx.partition, 0n) // Seek to beginning
		consumer.resume([{ topic: ctx.topic, partition: ctx.partition }])
	}
})
```

::: warning
Seeking does not affect committed offsets. If you want to persist the new position, you'll need to commit after seeking. On restart, the consumer will resume from the last committed offset.
:::

## Consumer Events

Listen to consumer lifecycle events:

```typescript
consumer.on('running', () => {
	console.log('Consumer started')
})

consumer.on('stopped', () => {
	console.log('Consumer stopped')
})

consumer.on('partitionsAssigned', partitions => {
	console.log('Assigned:', partitions)
})

consumer.on('partitionsRevoked', partitions => {
	console.log('Revoked:', partitions)
})

consumer.on('error', error => {
	console.error('Consumer error:', error)
})
```

## Static Membership

Use static membership to avoid rebalances during restarts:

```typescript
const consumer = client.consumer({
	groupId: 'my-group',
	groupInstanceId: 'instance-1', // Unique per consumer
	sessionTimeoutMs: 30000,
})
```

With static membership:

* Consumer keeps its partition assignment on restart
* No rebalance triggered if consumer rejoins within session timeout

## Manual Topic Assignment

Bypass the consumer group protocol entirely and manually assign specific partitions:

```typescript
interface ManualAssignment {
	topic: string
	partition: number
	offset?: bigint // Optional starting offset
}
```

### Basic Usage

```typescript
await consumer.runEach(
	'events',
	async message => {
		console.log(message.value)
	},
	{
		assignment: [
			{ topic: 'events', partition: 0 },
			{ topic: 'events', partition: 1 },
		],
	}
)
```

### With Explicit Starting Offset

```typescript
await consumer.runEach(
	'events',
	async message => {
		console.log(message.value)
	},
	{
		assignment: [{ topic: 'events', partition: 0, offset: 100n }],
	}
)
```

### How It Works

When using manual assignment:

* **No group coordination** - Skips JoinGroup/SyncGroup protocol exchanges
* **No rebalancing** - Partitions remain statically assigned
* **Offset resolution** - If `offset` is omitted, the consumer fetches committed offsets (if present) or falls back to `autoOffsetReset`
* **Offset commits** - Still commits offsets under `groupId` (if `commitOffsets: true`), but without group generation/member metadata

### Use Cases

* **Static partition mapping** - When you need deterministic partition-to-consumer mapping
* **Multiple consumers on same partition** - Multiple consumers with the same `groupId` can consume the same partition independently
* **Testing** - Simplified testing without rebalance complexity
* **State restoration** - Read specific partitions for state recovery

::: warning
Manual assignment does not emit `partitionsRevoked` events since there is no rebalancing. Ensure your application handles partition ownership appropriately.
:::

## Graceful Shutdown

Stop the consumer gracefully:

```typescript
const run = consumer.runEach('events', handler)

// Later: stop consuming
consumer.stop()
await run
```

With abort signal:

```typescript
const controller = new AbortController()

// Start consuming
const run = consumer.runEach('events', handler, { signal: controller.signal })

// Later: stop gracefully
controller.abort()
await run
```

## Partition Assignment Strategies

```typescript
const consumer = client.consumer({
	groupId: 'my-group',
	partitionAssignmentStrategy: 'cooperative-sticky', // Default
})
```

| Strategy               | Description                                            |
| ---------------------- | ------------------------------------------------------ |
| `'cooperative-sticky'` | Incremental rebalance (Kafka 2.4+), minimizes movement |
| `'sticky'`             | Minimize movement, eager rebalance                     |
| `'range'`              | Simple per-topic assignment                            |

## Isolation Level

Control visibility of transactional messages:

```typescript
const consumer = client.consumer({
	groupId: 'my-group',
	isolationLevel: 'read_committed', // Default - only committed transactions
	// isolationLevel: 'read_uncommitted', // See all messages
})
```

## Next Steps

* [Error Handling](/client/errors) - Error types and recovery
* [Codecs](/client/codecs) - Custom serialization
* [Transactions](/client/transactions) - Exactly-once consume-transform-produce

---

---
url: https://chrisrecalis.github.io/kafkats/client/share-consumer.md
---
# ShareConsumer API

The share consumer reads records from Kafka topics using Kafka **Share Groups** (KIP-932) for queue-like consumption with per-record acknowledgements.

`ShareConsumer` requires the **production-ready Kafka 4.2** Share Group APIs: ShareFetch and ShareAcknowledge v2.
These APIs include the KIP-1206 acquire modes and KIP-1222 acquisition-lock renewal. Kafka 4.2.1+ is recommended
because it fixes a critical Share Group broker deadlock. The preview APIs from Kafka 4.0 and 4.1 are not supported.

## Requirements

* Kafka 4.2+ with ShareFetch and ShareAcknowledge v2 is required
* Kafka 4.2.1+ is recommended
* Clusters running the latest production metadata version enable `share.version=1` by default

If the broker does not support Share APIs, `ShareConsumer` throws `KafkaFeatureUnsupportedError('share-groups')`.
If Share Groups are supported but disabled, it throws `KafkaFeatureDisabledError('share-groups')`.

## Creating a ShareConsumer

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})

const shareConsumer = client.shareConsumer({
	groupId: 'my-share-group',
})
```

## Subscriptions

`runEach()` and `stream()` take a `subscription` argument.

| Subscription               | What it does                       | Example                                                        |
| -------------------------- | ---------------------------------- | -------------------------------------------------------------- |
| Topic name                 | Consume raw `Buffer` key/value     | `shareConsumer.runEach('events', handler)`                     |
| Multiple topics            | Consume multiple topics at once    | `shareConsumer.runEach(['events', 'logs'], handler)`           |
| Typed topic (`topic(...)`) | Decode with codecs and infer types | `shareConsumer.runEach(userEvents, handler)`                   |
| Custom subscription        | Provide explicit decoders          | `shareConsumer.runEach({ topic: 'events', decoder }, handler)` |

## Processing Messages

### Single Message Mode (runEach)

```typescript
await shareConsumer.runEach('events', async (message, ctx) => {
	console.log({
		topic: ctx.topic,
		partition: ctx.partition,
		offset: ctx.offset,
		value: message.value?.toString('utf-8'),
	})
})
```

### Acknowledgements

`ShareConsumer` uses per-record acknowledgements:

* `message.ack()` (ACCEPT) — finalize successful processing
* `message.release()` (RELEASE) — return to the queue for redelivery
* `message.reject()` (REJECT) — drop without redelivery
* `message.renew()` (RENEW) — extend the acquisition lock without finalizing

`ack`, `release`, and `reject` each finalize the record; calling more than one for the same message throws. `renew` does **not** finalize — call it any number of times to keep the lock alive while a slow handler runs, then call one of the finalizing methods (or rely on `runEach()`'s implicit ACCEPT on success).

If your `runEach()` handler completes successfully and you did not call any of these methods, the record is **implicitly acknowledged** (ACCEPT).

In `stream()` mode, the record is **implicitly acknowledged** (ACCEPT) when you advance the iterator (or when the stream closes) unless you called `release()`/`reject()`.

```typescript
await shareConsumer.runEach('events', async message => {
	try {
		if (message.value !== null) {
			await process(message.value.toString('utf-8'))
		}
		await message.ack()
	} catch {
		await message.release()
	}
})
```

#### Renewing the acquisition lock (KIP-1222)

For long-running handlers that need more time than the broker's `group.share.record.lock.duration.ms`, call `message.renew()` to extend the lock:

```typescript
await shareConsumer.runEach('events', async (message, ctx) => {
	const heartbeat = setInterval(() => {
		message.renew().catch(err => ctx.signal.aborted || console.error(err))
	}, 20_000)
	try {
		if (message.value !== null) {
			await longRunningProcessing(message.value)
		}
	} finally {
		clearInterval(heartbeat)
	}
})
```

### Async Iterator Mode (stream)

Consume messages via `for await ... of`:

```typescript
for await (const { message, ctx } of shareConsumer.stream('events')) {
	console.log(ctx.topic, ctx.partition, ctx.offset, message.value?.toString('utf-8'))
	// Optionally:
	// await message.release()
	// await message.reject()
}
```

## Message Structure

```typescript
interface ShareMessage<V = Buffer, K = Buffer> {
	topic: string
	partition: number
	offset: bigint
	timestamp: bigint
	key: K | null
	value: V | null
	headers: Record<string, Buffer>
	deliveryCount?: number
	ack(): Promise<void>
	release(): Promise<void>
	reject(): Promise<void>
	renew(): Promise<void>
}
```

## Consume Context

```typescript
interface ConsumeContext {
	signal: AbortSignal
	topic: string
	partition: number
	offset: bigint
}
```

Use the signal to cancel long-running operations:

```typescript
await shareConsumer.runEach('events', async (message, ctx) => {
	const response = await fetch(url, { signal: ctx.signal })
	// ...
})
```

## Typed ShareConsumers

Use typed topics for automatic deserialization:

```typescript
import { topic, string, json } from '@kafkats/client'

interface UserEvent {
	userId: string
	action: string
}

const userEvents = topic('user-events', {
	key: string(),
	value: json<UserEvent>(),
})

await shareConsumer.runEach(userEvents, async message => {
	// message.key: string
	// message.value: UserEvent
	if (message.value !== null) {
		console.log(`User ${message.value.userId}: ${message.value.action}`)
	}
})
```

## Consumer Config

| Option        | Type                                  | Default             | Description                                                                                                            |
| ------------- | ------------------------------------- | ------------------- | ---------------------------------------------------------------------------------------------------------------------- |
| `groupId`     | `string`                              | required            | Share group identifier                                                                                                 |
| `rackId`      | `string`                              | -                   | Rack ID hint for rack-aware assignment                                                                                 |
| `maxWaitMs`   | `number`                              | `5000`              | `ShareFetch.max_wait_ms`                                                                                               |
| `minBytes`    | `number`                              | `1`                 | `ShareFetch.min_bytes`                                                                                                 |
| `maxBytes`    | `number`                              | `1048576`           | `ShareFetch.max_bytes`                                                                                                 |
| `maxRecords`  | `number`                              | `500`               | Max records the broker should return per fetch                                                                         |
| `batchSize`   | `number`                              | `100`               | Suggested batch size for acquired records and acknowledgements                                                         |
| `acquireMode` | `'batch_optimized' \| 'record_limit'` | `'batch_optimized'` | KIP-1206. `record_limit` strictly caps each fetch at `maxRecords`; `batch_optimized` may exceed it for batch alignment |

Against brokers without ShareFetch or ShareAcknowledge v2, the consumer throws `KafkaFeatureUnsupportedError`
before acquiring records.

## Run Options

### runEach Options

| Option          | Type          | Default | Description                                       |
| --------------- | ------------- | ------- | ------------------------------------------------- |
| `concurrency`   | `number`      | `10`    | Max records processed simultaneously              |
| `ackBatchSize`  | `number`      | `1000`  | Flush acknowledgements when this many are pending |
| `idleBackoffMs` | `number`      | `200`   | Backoff when no records are returned              |
| `signal`        | `AbortSignal` | -       | Abort to stop the consumer                        |

::: details ackBatchSize tuning
Acknowledgements are batched and coalesced to reduce network overhead. Consecutive offsets with the same acknowledgement type are combined into ranges before sending.

* **Lower values** (e.g., 100): Acks sent sooner, reducing risk of lock timeout for slow handlers, but more network requests
* **Higher values** (e.g., 2000): Fewer requests, but acks delayed longer

The default (1000) works well for most workloads.
:::

## Concurrency

Control how many records are processed concurrently:

```typescript
await shareConsumer.runEach('events', handler, { concurrency: 10 })
```

::: warning
Share Groups are queue-like: ordering is not guaranteed, and records from a single partition may be delivered/processed out of order (for example when redeliveries occur).
:::

## Graceful Shutdown

Stop the consumer gracefully:

```typescript
const run = shareConsumer.runEach('events', handler)

// Later: stop consuming
await shareConsumer.stop()
await run
```

`stop()` waits for shutdown cleanup to complete, including flushing any pending acknowledgements and leaving the share group.

With abort signal:

```typescript
const controller = new AbortController()
const run = shareConsumer.runEach('events', handler, { signal: controller.signal })

// Later: stop consuming
controller.abort()
await run
```

## Starting Position (latest)

Share Groups are queue-like: when a member starts fetching, it effectively begins from "now" (the current end offset).

If you want to ensure you only process records produced after your consumer is ready, wait for `partitionsAssigned` before producing:

```typescript
shareConsumer.on('partitionsAssigned', () => {
	// Safe point to produce messages you expect this ShareConsumer to receive.
})
```

## ShareConsumer Events

Listen to share consumer lifecycle events:

```typescript
shareConsumer.on('running', () => {
	console.log('ShareConsumer started')
})

shareConsumer.on('stopped', () => {
	console.log('ShareConsumer stopped')
})

shareConsumer.on('partitionsAssigned', partitions => {
	console.log('Assigned:', partitions)
})

shareConsumer.on('partitionsRevoked', partitions => {
	console.log('Revoked:', partitions)
})

shareConsumer.on('error', error => {
	console.error('ShareConsumer error:', error)
})
```

## When to Use Share Groups

Share Groups decouple parallelism from partition count. They excel when:

* **Message processing takes time** (10ms+): With regular consumer groups, concurrency is limited to partition count. Share Groups can process many messages concurrently regardless of partitions.
* **You need queue-like semantics**: Messages are delivered to any available consumer, not tied to specific partitions.
* **Ordering is not required**: Share Groups do not guarantee order.

### Performance Comparison

With 3 partitions, 20ms processing time per message, and 20 concurrent handlers:

| Consumer Type    | Throughput | Why                                         |
| ---------------- | ---------- | ------------------------------------------- |
| Regular Consumer | ~50 msg/s  | Limited to 3 concurrent (one per partition) |
| Share Consumer   | ~650 msg/s | 20 concurrent handlers                      |

For trivial handlers (0ms processing), regular consumers may be faster due to lower protocol overhead.

## Limitations

* No dead-letter queue / retry policy helpers (KIP-1191 lands in Kafka 4.4 and is not yet implemented)
* Share Groups do not support static membership
* Requires Kafka 4.2+ ShareFetch and ShareAcknowledge v2

---

---
url: https://chrisrecalis.github.io/kafkats/client/admin.md
---
# Admin API

The admin client provides cluster management operations for topics, consumer groups, and cluster metadata.

## Creating an Admin Client

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})

await client.connect()

const admin = client.admin()
```

The admin client shares the underlying cluster connection with the `KafkaClient`, so no additional connections are created.

## Topic Operations

### Fetching Topic Offsets

Get the earliest or latest offsets for a topic's partitions:

```typescript
const offsets = await admin.fetchTopicOffsets('events', [0, 1, 2], 'latest')

for (const [partition, offset] of offsets) {
	console.log(`Partition ${partition}: offset ${offset}`)
}
```

Fetch earliest offsets:

```typescript
const earliestOffsets = await admin.fetchTopicOffsets('events', [0, 1, 2], 'earliest')
```

With isolation level for transactional topics:

```typescript
// Get the last stable offset (LSO) - only committed transactional messages
const committedOffsets = await admin.fetchTopicOffsets('events', [0, 1, 2], 'latest', {
	isolationLevel: 'read_committed',
})
```

#### Options

| Option           | Type                                     | Default            | Description                                   |
| ---------------- | ---------------------------------------- | ------------------ | --------------------------------------------- |
| `topic`          | `string`                                 | -                  | Topic name                                    |
| `partitions`     | `number[]`                               | -                  | Partition indices to fetch offsets for        |
| `which`          | `'earliest' \| 'latest'`                 | -                  | Which offset to fetch                         |
| `isolationLevel` | `'read_uncommitted' \| 'read_committed'` | `read_uncommitted` | Controls visibility of transactional messages |

### Listing Topics

Get a list of all topic names in the cluster:

```typescript
const topics = await admin.listTopics()
console.log(topics) // ['orders', 'events', 'logs', ...]
```

### Describing Topics

Get detailed metadata for specific topics:

```typescript
const descriptions = await admin.describeTopics(['orders', 'events'])

for (const topic of descriptions) {
	console.log({
		name: topic.name,
		topicId: topic.topicId,
		isInternal: topic.isInternal,
		partitionCount: topic.partitions.length,
	})

	for (const partition of topic.partitions) {
		console.log({
			partition: partition.partitionIndex,
			leader: partition.leaderId,
			replicas: partition.replicas,
			isr: partition.isr,
		})
	}
}
```

Describe all topics by omitting the argument:

```typescript
const allTopics = await admin.describeTopics()
```

### Topic Description Structure

```typescript
interface TopicDescription {
	name: string // Topic name
	topicId: string // Topic UUID
	isInternal: boolean // Internal Kafka topic
	partitions: PartitionInfo[]
}

interface PartitionInfo {
	partitionIndex: number // Partition number
	leaderId: number // Leader broker ID
	leaderEpoch: number // Leader epoch
	replicas: number[] // Replica broker IDs
	isr: number[] // In-sync replica IDs
	offlineReplicas: number[] // Offline replica IDs
}
```

### Creating Topics

Create topics in the cluster:

```typescript
const results = await admin.createTopics([
	{
		name: 'orders',
		numPartitions: 6,
		replicationFactor: 3,
		configs: {
			'retention.ms': '604800000', // 7 days
			'cleanup.policy': 'delete',
		},
	},
])

for (const result of results) {
	if (result.errorCode === 0) {
		console.log(`Created topic: ${result.name}`)
	} else {
		console.log(`Failed to create ${result.name}: ${result.errorMessage}`)
	}
}
```

Validate topic configuration without creating:

```typescript
const results = await admin.createTopics([{ name: 'test-topic', numPartitions: 3 }], { validateOnly: true })
```

### Deleting Topics

Delete topics from the cluster:

```typescript
const results = await admin.deleteTopics(['old-topic', 'unused-topic'])

for (const result of results) {
	if (result.errorCode === 0) {
		console.log(`Deleted topic: ${result.name}`)
	} else {
		console.log(`Failed to delete ${result.name}: ${result.errorMessage}`)
	}
}
```

::: warning
Topic deletion is irreversible. All data in the topic will be lost.
:::

## Consumer Group Operations

### Listing Consumer Groups

List all consumer groups in the cluster:

```typescript
const groups = await admin.listGroups()

for (const group of groups) {
	console.log({
		groupId: group.groupId,
		protocolType: group.protocolType, // 'consumer'
		state: group.state, // 'Stable', 'Empty', etc.
	})
}
```

Filter groups by state:

```typescript
const stableGroups = await admin.listGroups({
	statesFilter: ['Stable'],
})
```

### Describing Consumer Groups

Get detailed information about consumer groups:

```typescript
const descriptions = await admin.describeGroups(['my-consumer-group'])

for (const group of descriptions) {
	console.log({
		groupId: group.groupId,
		state: group.state,
		protocol: group.protocol,
		members: group.members.length,
	})

	for (const member of group.members) {
		console.log({
			memberId: member.memberId,
			clientId: member.clientId,
			clientHost: member.clientHost,
			assignment: member.assignment, // Topic-partitions
		})
	}
}
```

### Consumer Group Description Structure

```typescript
interface ConsumerGroupDescription {
	groupId: string // Group ID
	state: string // 'Stable', 'Empty', 'Dead', etc.
	protocolType: string // Usually 'consumer'
	protocol: string // Assignment strategy name
	members: MemberDescription[]
	errorCode: number // 0 if successful
}

interface MemberDescription {
	memberId: string // Unique member ID
	groupInstanceId: string | null // Static membership ID
	clientId: string // Client identifier
	clientHost: string // Member host
	assignment: TopicPartition[] // Assigned partitions
}

interface TopicPartition {
	topic: string
	partition: number
}
```

### Deleting Consumer Groups

Delete consumer groups that have no active members:

```typescript
const results = await admin.deleteGroups(['old-group'])

for (const result of results) {
	if (result.errorCode === 0) {
		console.log(`Deleted group: ${result.groupId}`)
	} else {
		console.log(`Failed to delete ${result.groupId}`)
	}
}
```

::: warning
Consumer groups can only be deleted when they have no active members. Attempting to delete a non-empty group returns `NonEmptyGroup` error.
:::

## Cluster Operations

### Describing the Cluster

Get cluster metadata including broker information:

```typescript
const cluster = await admin.describeCluster()

console.log({
	clusterId: cluster.clusterId,
	controllerId: cluster.controllerId,
	brokerCount: cluster.brokers.length,
})

for (const broker of cluster.brokers) {
	console.log({
		nodeId: broker.nodeId,
		host: broker.host,
		port: broker.port,
		rack: broker.rack,
	})
}
```

### Cluster Description Structure

```typescript
interface ClusterDescription {
	clusterId: string | null // Cluster identifier
	controllerId: number // Controller broker ID
	brokers: BrokerDescription[]
}

interface BrokerDescription {
	nodeId: number // Broker ID
	host: string // Broker host
	port: number // Broker port
	rack: string | null // Rack identifier
}
```

## ACL Operations

Access Control Lists (ACLs) control which users can perform which operations on which resources.

### Describing ACLs

Query ACLs matching a filter:

```typescript
import { AclResourceType, AclResourcePatternType, AclOperation, AclPermissionType } from '@kafkats/client'

// Describe all ACLs for a specific topic
const result = await admin.describeAcls({
	resourceTypeFilter: AclResourceType.TOPIC,
	resourceNameFilter: 'my-topic',
	patternTypeFilter: AclResourcePatternType.LITERAL,
	principalFilter: null, // null matches any
	hostFilter: null,
	operation: AclOperation.ANY,
	permissionType: AclPermissionType.ANY,
})

for (const resource of result.resources) {
	console.log(`${resource.resourceType}: ${resource.resourceName}`)
	for (const acl of resource.acls) {
		console.log(`  ${acl.principal} ${acl.permissionType} ${acl.operation}`)
	}
}
```

Describe all ACLs in the cluster:

```typescript
const result = await admin.describeAcls({
	resourceTypeFilter: AclResourceType.ANY,
	resourceNameFilter: null,
	patternTypeFilter: AclResourcePatternType.ANY,
	principalFilter: null,
	hostFilter: null,
	operation: AclOperation.ANY,
	permissionType: AclPermissionType.ANY,
})
```

### Creating ACLs

Create ACL bindings to grant or deny access:

```typescript
// Allow User:alice to read from my-topic
const results = await admin.createAcls([
	{
		resourceType: AclResourceType.TOPIC,
		resourceName: 'my-topic',
		resourcePatternType: AclResourcePatternType.LITERAL,
		principal: 'User:alice',
		host: '*',
		operation: AclOperation.READ,
		permissionType: AclPermissionType.ALLOW,
	},
])

for (const result of results) {
	if (result.errorCode === 0) {
		console.log('ACL created successfully')
	} else {
		console.log(`Failed: ${result.errorMessage}`)
	}
}
```

Create prefixed ACLs to match multiple resources:

```typescript
// Allow User:bob to write to all topics starting with "events-"
await admin.createAcls([
	{
		resourceType: AclResourceType.TOPIC,
		resourceName: 'events-',
		resourcePatternType: AclResourcePatternType.PREFIXED,
		principal: 'User:bob',
		host: '*',
		operation: AclOperation.WRITE,
		permissionType: AclPermissionType.ALLOW,
	},
])
```

### Deleting ACLs

Delete ACLs matching a filter:

```typescript
// Delete all ACLs for User:alice on my-topic
const results = await admin.deleteAcls([
	{
		resourceTypeFilter: AclResourceType.TOPIC,
		resourceNameFilter: 'my-topic',
		patternTypeFilter: AclResourcePatternType.LITERAL,
		principalFilter: 'User:alice',
		hostFilter: null,
		operation: AclOperation.ANY,
		permissionType: AclPermissionType.ANY,
	},
])

for (const result of results) {
	console.log(`Deleted ${result.matchingAcls.length} ACLs`)
}
```

### ACL Types

#### Resource Types

| Type               | Description              |
| ------------------ | ------------------------ |
| `TOPIC`            | Topic resource           |
| `GROUP`            | Consumer group           |
| `CLUSTER`          | Cluster-level operations |
| `TRANSACTIONAL_ID` | Transactional ID         |
| `DELEGATION_TOKEN` | Delegation token         |
| `ANY`              | Match any (for filters)  |

#### Operations

| Operation          | Description             |
| ------------------ | ----------------------- |
| `READ`             | Read from resource      |
| `WRITE`            | Write to resource       |
| `CREATE`           | Create resource         |
| `DELETE`           | Delete resource         |
| `ALTER`            | Alter resource          |
| `DESCRIBE`         | Describe resource       |
| `CLUSTER_ACTION`   | Cluster actions         |
| `DESCRIBE_CONFIGS` | Describe configs        |
| `ALTER_CONFIGS`    | Alter configs           |
| `IDEMPOTENT_WRITE` | Idempotent writes       |
| `ALL`              | All operations          |
| `ANY`              | Match any (for filters) |

#### Pattern Types

| Type       | Description                |
| ---------- | -------------------------- |
| `LITERAL`  | Exact resource name match  |
| `PREFIXED` | Resource name prefix match |
| `ANY`      | Match any (for filters)    |

#### Permission Types

| Type    | Description             |
| ------- | ----------------------- |
| `ALLOW` | Allow the operation     |
| `DENY`  | Deny the operation      |
| `ANY`   | Match any (for filters) |

## Error Handling

Admin operations return results with error codes for each item:

```typescript
import { ErrorCode } from '@kafkats/client'

const results = await admin.deleteTopics(['my-topic'])

for (const result of results) {
	switch (result.errorCode) {
		case ErrorCode.None:
			console.log(`Success: ${result.name}`)
			break
		case ErrorCode.UnknownTopicOrPartition:
			console.log(`Topic not found: ${result.name}`)
			break
		case ErrorCode.TopicAuthorizationFailed:
			console.log(`Not authorized: ${result.name}`)
			break
		default:
			console.log(`Error ${result.errorCode}: ${result.errorMessage}`)
	}
}
```

### Common Error Codes

| Error Code                   | Description                          |
| ---------------------------- | ------------------------------------ |
| `None` (0)                   | Operation succeeded                  |
| `UnknownTopicOrPartition`    | Topic does not exist                 |
| `TopicAlreadyExists`         | Topic already exists (create)        |
| `NonEmptyGroup`              | Group has active members (delete)    |
| `GroupIdNotFound`            | Group does not exist                 |
| `TopicAuthorizationFailed`   | Not authorized for topic operation   |
| `GroupAuthorizationFailed`   | Not authorized for group operation   |
| `ClusterAuthorizationFailed` | Not authorized for cluster operation |

## Admin Options

Configure admin behavior:

```typescript
const admin = client.admin({
	requestTimeoutMs: 30000, // Timeout for admin operations (default: 30s)
})
```

| Option             | Type     | Default | Description                     |
| ------------------ | -------- | ------- | ------------------------------- |
| `requestTimeoutMs` | `number` | `30000` | Timeout for admin requests (ms) |

## Next Steps

* [Error Handling](/client/errors) - Error types and recovery
* [Cluster API](/client/advanced/cluster) - Low-level cluster operations
* [Configuration](/client/configuration) - Full configuration reference

---

---
url: https://chrisrecalis.github.io/kafkats/client/compression.md
---
# Compression

kafkats supports multiple compression algorithms for reducing network bandwidth and storage. Compression is applied at the RecordBatch level - the producer compresses batches before sending, and consumers automatically decompress.

## Quick Start

Install a supported compression library and use it — kafkats detects and registers it automatically:

```bash
npm install snappy
```

```typescript
const producer = client.producer({
	compression: 'snappy',
})
```

No registration call is needed. When a codec is first looked up, kafkats checks for the supported libraries (see below) and registers the first one it finds.

## Compression Types

| Type       | Speed     | Ratio | Built-in | Notes                                       |
| ---------- | --------- | ----- | -------- | ------------------------------------------- |
| `'none'`   | Fastest   | 1:1   | Yes      | No compression                              |
| `'gzip'`   | Slow      | Best  | Yes      | Uses Node.js zlib                           |
| `'snappy'` | Fast      | Good  | No       | Balanced choice, auto-detected library      |
| `'lz4'`    | Very fast | Good  | No       | Best for throughput, auto-detected library  |
| `'zstd'`   | Medium    | Best  | No       | Modern and efficient, auto-detected library |

## Built-in Codecs

GZIP is built-in and requires no additional setup:

```typescript
const producer = client.producer({
	compression: 'gzip',
})
```

## Automatic Codec Registration

For Snappy, LZ4, and Zstd, install one of the supported libraries and kafkats picks it up automatically — no registration code required. When several are installed, the first match in the table below (fastest first) wins.

### Snappy

| Library    | Type    | Performance | Auto-detected |
| ---------- | ------- | ----------- | ------------- |
| `snappy`   | Native  | Fastest     | Yes (1st)     |
| `snappyjs` | Pure JS | Good        | Yes (2nd)     |

```bash
npm install snappy
```

### LZ4

| Library    | Type    | Performance | Auto-detected |
| ---------- | ------- | ----------- | ------------- |
| `lz4-napi` | Native  | Fastest     | Yes (1st)     |
| `lz4`      | Native  | Fast        | Yes (2nd)     |
| `lz4js`    | Pure JS | Good        | Yes (3rd)     |

```bash
npm install lz4-napi
```

::: warning
`lz4-napi` 2.x or later is required — Kafka needs the LZ4 frame format, which older versions don't expose.
:::

### Zstd

| Library            | Type   | Performance | Auto-detected |
| ------------------ | ------ | ----------- | ------------- |
| `@mongodb-js/zstd` | Native | Fastest     | Yes (1st)     |
| `zstd-napi`        | Native | Fastest     | Yes (2nd)     |
| `zstd-codec`       | WASM   | Good        | No (manual)   |

```bash
npm install @mongodb-js/zstd
```

::: warning
`@mongodb-js/zstd` v7+ requires Node 20.19 or later. On Node 18, install `@mongodb-js/zstd@2` (or `zstd-napi`) instead.
:::

### Disabling auto-registration

If you want full control over which codecs are used, turn auto-registration off and register codecs explicitly:

```typescript
import { compressionCodecs } from '@kafkats/client'

compressionCodecs.autoRegister = false
```

## Manual Registration

Manual registration is still available — it always takes precedence over auto-detection. Use it for custom codecs, for `zstd-codec` (which needs async initialization), or to pass options like the Zstd compression level.

### Snappy

```typescript
import snappy from 'snappy' // or: import * as SnappyJS from 'snappyjs'
import { CompressionType, compressionCodecs, createSnappyCodec } from '@kafkats/client'

compressionCodecs.register(CompressionType.Snappy, createSnappyCodec(snappy))
```

### LZ4

```typescript
import * as lz4 from 'lz4-napi' // or: 'lz4', 'lz4js'
import { CompressionType, compressionCodecs, createLz4Codec } from '@kafkats/client'

compressionCodecs.register(CompressionType.Lz4, createLz4Codec(lz4))
```

### Zstd

```typescript
import { compress, decompress } from '@mongodb-js/zstd' // or: 'zstd-napi'
import { CompressionType, compressionCodecs, createZstdCodec } from '@kafkats/client'

compressionCodecs.register(CompressionType.Zstd, createZstdCodec({ compress, decompress }))
```

#### zstd-codec (WASM, manual only)

`zstd-codec` initializes asynchronously, so it cannot be auto-detected and must be registered manually:

```bash
npm install zstd-codec
```

```typescript
import { ZstdCodec } from 'zstd-codec'
import { CompressionType, compressionCodecs, createZstdCodec } from '@kafkats/client'

// Initialize and register within callback
ZstdCodec.run(zstd => {
	const simple = new zstd.Simple()
	compressionCodecs.register(CompressionType.Zstd, createZstdCodec(simple))
})
```

## Compression Options

### Zstd Compression Level

Zstd supports compression levels from 1-22 (default: 3). Lower levels are faster, higher levels achieve better compression:

```typescript
import { compress, decompress } from '@mongodb-js/zstd'

compressionCodecs.register(CompressionType.Zstd, createZstdCodec({ compress, decompress }, { level: 6 }))
```

## Transparent Decompression

Consumers automatically detect and decompress messages without any configuration. The compression type is stored in the RecordBatch header, so consumers can decode messages regardless of which compression was used by the producer.

```typescript
// Producer uses gzip compression (built-in)
const producer = client.producer({ compression: 'gzip' })
await producer.send('my-topic', [{ value: Buffer.from('compressed data') }])

// Consumer automatically decompresses
const consumer = client.consumer({ groupId: 'my-group' })
for await (const { message } of consumer.stream('my-topic')) {
	console.log(message.value.toString()) // 'compressed data'
}
```

::: tip
Make sure a compression library for the topic's compression type is installed (or a codec manually registered) before consuming. GZIP works out of the box; Snappy/LZ4/Zstd need one of the supported libraries installed.
:::

## Performance Considerations

Choose your compression strategy based on your use case:

| Use Case                 | Recommended  | Why                               |
| ------------------------ | ------------ | --------------------------------- |
| High throughput, low CPU | LZ4 or None  | Fastest compression/decompression |
| Network-constrained      | Zstd or Gzip | Best compression ratio            |
| Balanced workload        | Snappy       | Good mix of speed and compression |
| Log/text data            | Gzip or Zstd | Text compresses well with these   |

## Supported Libraries Summary

### Snappy

* **Native**: [`snappy`](https://www.npmjs.com/package/snappy) - Fastest, napi-rs based
* **Pure JS**: [`snappyjs`](https://www.npmjs.com/package/snappyjs)

### LZ4

* **Native**: [`lz4-napi`](https://www.npmjs.com/package/lz4-napi) - Fastest, napi-rs based
* **Native**: [`lz4`](https://www.npmjs.com/package/lz4) - node-lz4, encode/decode API
* **Pure JS**: [`lz4js`](https://www.npmjs.com/package/lz4js)

### Zstd

* **Native**: [`@mongodb-js/zstd`](https://www.npmjs.com/package/@mongodb-js/zstd) - MongoDB's binding
* **Native**: [`zstd-napi`](https://www.npmjs.com/package/zstd-napi) - Node-API binding
* **WASM**: [`zstd-codec`](https://www.npmjs.com/package/zstd-codec) - Emscripten based

## Custom Codecs

You can also implement your own compression codec:

```typescript
import { CompressionCodec, CompressionType, compressionCodecs } from '@kafkats/client'

const myCodec: CompressionCodec = {
	async compress(data: Buffer): Promise<Buffer> {
		// Your compression logic
		return compressedData
	},
	async decompress(data: Buffer): Promise<Buffer> {
		// Your decompression logic
		return decompressedData
	},
}

compressionCodecs.register(CompressionType.Snappy, myCodec)
```

## Next Steps

* [Producer API](/client/producer) - Configure producer compression
* [Configuration](/client/configuration) - Full configuration reference
* [Codecs](/client/codecs) - Message serialization (different from compression)

---

---
url: https://chrisrecalis.github.io/kafkats/client/codecs.md
---
# Codecs

Codecs handle serialization and deserialization of message keys and values. kafkats provides built-in codecs and supports custom implementations.

## Built-in Codecs

### String Codec

Encodes/decodes UTF-8 strings:

```typescript
import { string } from '@kafkats/client'

const codec = string()
codec.encode('hello') // Buffer
codec.decode(buffer) // 'hello'
```

### JSON Codec

Encodes/decodes JSON with TypeScript generics:

```typescript
import { json } from '@kafkats/client'

interface User {
	id: string
	name: string
}

const codec = json<User>()
codec.encode({ id: '1', name: 'Alice' }) // Buffer (JSON string)
codec.decode(buffer) // { id: '1', name: 'Alice' }
```

### Buffer Codec

Passthrough for raw binary data:

```typescript
import { buffer } from '@kafkats/client'

const codec = buffer()
codec.encode(data) // Same Buffer
codec.decode(buf) // Same Buffer
```

## Using Codecs with Topics

Define typed topics with codecs:

```typescript
import { topic, string, json } from '@kafkats/client'

interface OrderEvent {
	orderId: string
	status: 'created' | 'shipped' | 'delivered'
}

const orders = topic('orders', {
	key: string(),
	value: json<OrderEvent>(),
})

// Type-safe producer
await producer.send(orders, [{ key: 'order-123', value: { orderId: 'order-123', status: 'created' } }])

// Type-safe consumer
await consumer.runEach(orders, async message => {
	// message.key: string
	// message.value: OrderEvent
})
```

## Custom Codecs

Create custom codecs for any serialization format:

```typescript
import { codec } from '@kafkats/client'

// Simple custom codec
const intCodec = codec<number>(
	n => {
		const buf = Buffer.alloc(4)
		buf.writeInt32BE(n)
		return buf
	},
	buf => buf.readInt32BE()
)
```

### Protocol Buffers Example

```typescript
import { codec } from '@kafkats/client'
import { User } from './generated/user_pb.js'

const userCodec = codec<User>(
	user => Buffer.from(user.serializeBinary()),
	buf => User.deserializeBinary(buf)
)

const users = topic('users', {
	key: string(),
	value: userCodec,
})
```

### Avro Example

```typescript
import { codec } from '@kafkats/client'
import avro from 'avsc'

const userType = avro.Type.forSchema({
	type: 'record',
	name: 'User',
	fields: [
		{ name: 'id', type: 'string' },
		{ name: 'name', type: 'string' },
	],
})

const avroCodec = codec<{ id: string; name: string }>(
	user => userType.toBuffer(user),
	buf => userType.fromBuffer(buf)
)
```

## Codec Interface

A codec must implement the `Codec<T>` interface:

```typescript
interface Codec<T> {
	encode(value: T): Buffer
	decode(buffer: Buffer): T
}
```

## Custom Value Codecs

For simpler cases, define a value codec inline:

```typescript
import { topic } from '@kafkats/client'

const events = topic('events', {
	value: {
		encode: (value: MyType) => Buffer.from(JSON.stringify(value)),
		decode: (buf: Buffer) => JSON.parse(buf.toString()) as MyType,
	},
})
```

## Key-Only or Value-Only Codecs

You can specify codecs for just keys or just values:

```typescript
// Key codec only (value stays as Buffer)
const keyed = topic('keyed', {
	key: string(),
})

// Value codec only (key stays as Buffer)
const valued = topic('valued', {
	value: json<MyEvent>(),
})
```

## Null Handling

Codecs receive/return `null` for missing values:

```typescript
const nullableCodec = codec<string | null>(
	value => (value === null ? Buffer.alloc(0) : Buffer.from(value)),
	buf => (buf.length === 0 ? null : buf.toString())
)
```

## Using with Zod

For runtime validation, use [@kafkats/flow-codec-zod](/flow-codec-zod/):

```typescript
import { zodCodec } from '@kafkats/flow-codec-zod'
import { z } from 'zod'

const UserSchema = z.object({
	id: z.string(),
	email: z.string().email(),
})

const userCodec = zodCodec(UserSchema)
// Validates on both encode and decode
```

## Performance Considerations

* **Reuse codecs** - Create codec instances once, reuse them
* **Buffer pooling** - For high-throughput, consider pooling buffers
* **Pre-size buffers** - Calculate buffer size upfront when possible

```typescript
// Efficient: pre-sized buffer
const efficientCodec = codec<number[]>(
	arr => {
		const buf = Buffer.alloc(4 * arr.length)
		arr.forEach((n, i) => buf.writeInt32BE(n, i * 4))
		return buf
	},
	buf => {
		const arr: number[] = []
		for (let i = 0; i < buf.length; i += 4) {
			arr.push(buf.readInt32BE(i))
		}
		return arr
	}
)
```

---

---
url: https://chrisrecalis.github.io/kafkats/client/authentication.md
---
# Authentication

kafkats supports SASL authentication for secure Kafka connections.

## Supported Mechanisms

| Mechanism       | Description                      |
| --------------- | -------------------------------- |
| `PLAIN`         | Username/password (use with TLS) |
| `SCRAM-SHA-256` | Challenge-response, SHA-256      |
| `SCRAM-SHA-512` | Challenge-response, SHA-512      |
| `OAUTHBEARER`   | Bearer token (e.g. AWS MSK IAM)  |

## SASL Options

SASL config is a discriminated union keyed by `mechanism`.

### PLAIN / SCRAM

| Option      | Type                                            | Required | Description         |
| ----------- | ----------------------------------------------- | -------- | ------------------- |
| `mechanism` | `'PLAIN' \| 'SCRAM-SHA-256' \| 'SCRAM-SHA-512'` | Yes      | SASL mechanism name |
| `username`  | `string`                                        | Yes      | SASL username       |
| `password`  | `string`                                        | Yes      | SASL password       |

### OAUTHBEARER

| Option                        | Type                                    | Required | Description                                                                                     |
| ----------------------------- | --------------------------------------- | -------- | ----------------------------------------------------------------------------------------------- |
| `mechanism`                   | `'OAUTHBEARER'`                         | Yes      | SASL mechanism name                                                                             |
| `oauthBearerProvider`         | `(context) => ({ value, extensions? })` | Yes      | Returns the bearer token (and optional extensions) for the broker                               |
| `reauthenticationThresholdMs` | `number`                                | No       | Reauthenticate when this many milliseconds remain of broker session lifetime (default: `10000`) |

`context` includes `{ host, port, clientId }`.

## SASL/PLAIN

Simple username/password authentication. Always use with TLS:

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: { enabled: true },
	sasl: {
		mechanism: 'PLAIN',
		username: 'my-username',
		password: 'my-password',
	},
})
```

::: warning Security
PLAIN mechanism sends credentials in base64 (not encrypted). Always use TLS (`tls: { enabled: true }`) to protect credentials in transit.
:::

## SCRAM-SHA-256

More secure challenge-response authentication:

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: { enabled: true },
	sasl: {
		mechanism: 'SCRAM-SHA-256',
		username: 'my-username',
		password: 'my-password',
	},
})
```

## SCRAM-SHA-512

Strongest built-in authentication:

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: { enabled: true },
	sasl: {
		mechanism: 'SCRAM-SHA-512',
		username: 'my-username',
		password: 'my-password',
	},
})
```

## OAUTHBEARER

Provide a bearer token per broker connection via `oauthBearerProvider`. Tokens are commonly short-lived, so generate them on demand and refresh when needed.

If the broker has periodic reauthentication enabled (`connections.max.reauth.ms`), kafkats will automatically reauthenticate on the existing connection using `SaslAuthenticate` before the session expires.

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: { enabled: true },
	sasl: {
		mechanism: 'OAUTHBEARER',
		oauthBearerProvider: async ({ host, port }) => {
			const value = await getTokenForBroker(`${host}:${port}`)
			return { value }
		},
	},
})
```

## TLS Configuration

### TLS Options

| Option               | Type                                          | Default | Description                  |
| -------------------- | --------------------------------------------- | ------- | ---------------------------- |
| `enabled`            | `boolean`                                     | `false` | Enables TLS when `true`      |
| `ca`                 | `string \| Buffer \| Array<string \| Buffer>` | -       | CA certificate(s)            |
| `cert`               | `string \| Buffer`                            | -       | Client certificate (mTLS)    |
| `key`                | `string \| Buffer`                            | -       | Client private key (mTLS)    |
| `passphrase`         | `string`                                      | -       | Private key passphrase       |
| `rejectUnauthorized` | `boolean`                                     | `true`  | Validate broker certificates |
| `servername`         | `string`                                      | -       | SNI server name              |

### Basic TLS

Use system CA certificates:

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: { enabled: true },
})
```

### Custom CA Certificate

```typescript
import { readFileSync } from 'fs'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: {
		enabled: true,
		ca: readFileSync('/path/to/ca.pem'),
	},
})
```

### Mutual TLS (mTLS)

Client certificate authentication:

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['kafka.example.com:9093'],
	tls: {
		enabled: true,
		ca: readFileSync('/path/to/ca.pem'),
		cert: readFileSync('/path/to/client.pem'),
		key: readFileSync('/path/to/client-key.pem'),
	},
})
```

### Disable Certificate Verification

::: danger Not for Production
Only use this for development/testing with self-signed certificates.
:::

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9093'],
	tls: {
		enabled: true,
		rejectUnauthorized: false,
	},
})
```

## Environment Variables

Common pattern for configuration:

```typescript
const client = new KafkaClient({
	clientId: process.env.KAFKA_CLIENT_ID || 'my-app',
	brokers: (process.env.KAFKA_BROKERS || 'localhost:9092').split(','),
	tls: process.env.KAFKA_TLS_ENABLED === 'true' ? { enabled: true } : undefined,
	sasl:
		process.env.KAFKA_SASL_MECHANISM === 'OAUTHBEARER'
			? {
					mechanism: 'OAUTHBEARER',
					oauthBearerProvider: async () => {
						// For Amazon MSK IAM, install the AWS signer:
						//   pnpm add aws-msk-iam-sasl-signer-js
						const { generateAuthToken } = await import('aws-msk-iam-sasl-signer-js')
						const { token } = await generateAuthToken({ region: process.env.AWS_REGION! })
						return { value: token }
					},
				}
			: process.env.KAFKA_SASL_USERNAME
				? {
						mechanism: (process.env.KAFKA_SASL_MECHANISM || 'SCRAM-SHA-256') as 'SCRAM-SHA-256',
						username: process.env.KAFKA_SASL_USERNAME,
						password: process.env.KAFKA_SASL_PASSWORD!,
					}
				: undefined,
})
```

## Broker Configuration

### Confluent Cloud

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['xxx.confluent.cloud:9092'],
	tls: { enabled: true },
	sasl: {
		mechanism: 'PLAIN',
		username: process.env.CONFLUENT_API_KEY!,
		password: process.env.CONFLUENT_API_SECRET!,
	},
})
```

### Amazon MSK (IAM)

Amazon MSK IAM can be used via SASL `OAUTHBEARER` by returning a SigV4-based token from `oauthBearerProvider`.

References:

* [AWS MSK Developer Guide: Configure clients for IAM access control](https://docs.aws.amazon.com/msk/latest/developerguide/configure-clients-for-iam-access-control.html)
* [AWS MSK IAM SASL signer for JavaScript (`aws-msk-iam-sasl-signer-js`)](https://github.com/aws/aws-msk-iam-sasl-signer-js#getting-started)

```typescript
import { KafkaClient } from '@kafkats/client'

async function createMskIamToken(options: { region: string }): Promise<string> {
	// Recommended: use AWS' official MSK IAM SASL signer (Node.js)
	// Install:
	//   pnpm add aws-msk-iam-sasl-signer-js
	// (AWS also documents installing from GitHub:
	//   npm install https://github.com/aws/aws-msk-iam-sasl-signer-js)
	// It also supports fetching creds from a profile or role:
	//   generateAuthTokenFromProfile({ region, awsProfileName })
	//   generateAuthTokenFromRole({ region, awsRoleArn, awsRoleSessionName? })
	const { generateAuthToken } = await import('aws-msk-iam-sasl-signer-js')
	const { token, expiryTime } = await generateAuthToken({ region: options.region })
	// expiryTime is milliseconds since epoch (useful if you want to cache/refresh)
	return token
}

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['b-1.msk.example.amazonaws.com:9098'],
	tls: { enabled: true },
	sasl: {
		mechanism: 'OAUTHBEARER',
		oauthBearerProvider: async () => ({ value: await createMskIamToken({ region: 'us-east-1' }) }),
	},
})
```

::: warning Token lifetime
MSK IAM tokens are short-lived. Generate them on demand inside `oauthBearerProvider` rather than hard-coding a static token.
:::

::: tip How the token is built
The AWS signer generates a SigV4 presigned URL for the `kafka-cluster` service with `Action=kafka-cluster:Connect`, then base64url-encodes it for use as the OAUTHBEARER token.
:::

### Redpanda

```typescript
const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['redpanda.example.com:9092'],
	tls: { enabled: true },
	sasl: {
		mechanism: 'SCRAM-SHA-256',
		username: 'user',
		password: 'password',
	},
})
```

## Troubleshooting

### Authentication Failed

```
Error: SASL authentication failed: Invalid credentials
```

* Verify username and password
* Check if the mechanism matches broker configuration
* Ensure the user has proper ACLs

### Connection Refused

```
Error: Connection refused
```

* Check if broker address is correct
* Verify TLS port (usually 9093) vs plaintext (9092)
* Check firewall rules

### Certificate Errors

```
Error: unable to verify the first certificate
```

* Add the CA certificate to your configuration
* Or set `rejectUnauthorized: false` for testing

---

---
url: https://chrisrecalis.github.io/kafkats/client/errors.md
---
# Error Handling

kafkats provides specific error types for different failure scenarios, making it easy to handle errors appropriately.

## Error Hierarchy

```
KafkaError (base)
├── KafkaProtocolError (protocol-level errors)
├── ConnectionError (network failures)
├── TimeoutError (request timeouts)
└── Specific errors...
```

## Common Errors

### ConnectionError

Network-level connection failures:

```typescript
import { ConnectionError } from '@kafkats/client'

try {
	await producer.send('events', [{ value: 'payload' }])
} catch (error) {
	if (error instanceof ConnectionError) {
		console.log('Failed to connect:', error.message)
		// Retry or fail gracefully
	}
}
```

### TimeoutError

Request took too long:

```typescript
import { TimeoutError } from '@kafkats/client'

try {
	await consumer.runEach('events', async () => {
		// ...
	})
} catch (error) {
	if (error instanceof TimeoutError) {
		console.log('Request timed out')
		// Increase timeout or retry
	}
}
```

### SendTimeoutError

Producer send timed out:

```typescript
import { SendTimeoutError } from '@kafkats/client'

try {
	await producer.send('events', { value: 'data' })
} catch (error) {
	if (error instanceof SendTimeoutError) {
		console.log('Send timed out after retries')
		// Message may or may not have been delivered
	}
}
```

### RecordTooLargeError

Message exceeds broker limits:

```typescript
import { RecordTooLargeError } from '@kafkats/client'

try {
	await producer.send('events', { value: hugePayload })
} catch (error) {
	if (error instanceof RecordTooLargeError) {
		console.log('Message too large:', error.message)
		// Split the message or increase broker limits
	}
}
```

## Broker Errors

### LeaderNotAvailableError

Partition leader is unavailable:

```typescript
import { LeaderNotAvailableError } from '@kafkats/client'

// Usually retriable - kafkats handles this automatically
```

### CoordinatorNotAvailableError

Group coordinator is unavailable:

```typescript
import { CoordinatorNotAvailableError } from '@kafkats/client'

// Consumer will retry finding the coordinator
```

### UnknownTopicOrPartitionError

Topic doesn't exist:

```typescript
import { UnknownTopicOrPartitionError } from '@kafkats/client'

try {
	await producer.send('nonexistent', { value: 'data' })
} catch (error) {
	if (error instanceof UnknownTopicOrPartitionError) {
		console.log('Topic not found:', error.message)
	}
}
```

## Consumer Group Errors

### RebalanceInProgressError

Consumer group is rebalancing:

```typescript
import { RebalanceInProgressError } from '@kafkats/client'

// Handled automatically - wait for rebalance to complete
```

### UnknownMemberIdError

Consumer was removed from group:

```typescript
import { UnknownMemberIdError } from '@kafkats/client'

// Consumer will rejoin the group
```

### IllegalGenerationError

Consumer has stale generation:

```typescript
import { IllegalGenerationError } from '@kafkats/client'

// Consumer will rejoin with new generation
```

## Checking Retriability

Use `isRetriable()` to check if an error can be retried:

```typescript
import { isRetriable } from '@kafkats/client'

const topic = 'events'
const messages = [{ value: 'data' }]

try {
	await producer.send(topic, messages)
} catch (error) {
	if (isRetriable(error)) {
		// Error is transient, retry might succeed
		await delay(1000)
		await producer.send(topic, messages)
	} else {
		// Permanent error, don't retry
		throw error
	}
}
```

## Error Utilities

### isKafkaError

Check if an error is from kafkats:

```typescript
import { isKafkaError } from '@kafkats/client'

const topic = 'events'
const messages = [{ value: 'data' }]

try {
	await producer.send(topic, messages)
} catch (error) {
	if (isKafkaError(error)) {
		console.log('Kafka error:', error.code, error.message)
	} else {
		console.log('Other error:', error)
	}
}
```

### shouldRefreshMetadata

Check if metadata should be refreshed:

```typescript
import { shouldRefreshMetadata } from '@kafkats/client'

const topic = 'events'
const messages = [{ value: 'data' }]

try {
	await producer.send(topic, messages)
} catch (error) {
	if (shouldRefreshMetadata(error)) {
		// Metadata might be stale, kafkats refreshes automatically
	}
}
```

## Producer Error Events

Listen for producer errors:

```typescript
producer.on('error', error => {
	console.error('Producer error:', error)
})
```

## Consumer Error Events

Listen for consumer errors:

```typescript
consumer.on('error', error => {
	console.error('Consumer error:', error)
})

// Session lost - partitions are no longer owned
consumer.on('partitionsLost', partitions => {
	console.log('Lost partitions:', partitions)
	// Cannot commit offsets for these partitions
})
```

## Error Handling Patterns

### Retry with Backoff

```typescript
async function sendWithRetry(producer: Producer, topic: string, messages: ProducerMessage[], maxRetries = 3) {
	for (let attempt = 0; attempt < maxRetries; attempt++) {
		try {
			return await producer.send(topic, messages)
		} catch (error) {
			if (!isRetriable(error) || attempt === maxRetries - 1) {
				throw error
			}
			await delay(100 * Math.pow(2, attempt))
		}
	}
}
```

### Dead Letter Queue

```typescript
await consumer.runEach('my-topic', async (message, ctx) => {
	try {
		await processMessage(message)
	} catch (error) {
		// Send failed messages to DLQ
		await dlqProducer.send('my-topic-dlq', [
			{
				key: message.key,
				value: message.value,
				headers: {
					error: error instanceof Error ? error.message : String(error),
					originalTopic: ctx.topic,
					originalPartition: String(ctx.partition),
				},
			},
		])
	}
})
```

### Circuit Breaker

```typescript
class CircuitBreaker {
	private failures = 0
	private lastFailure = 0
	private readonly threshold = 5
	private readonly resetMs = 30000

	async call<T>(fn: () => Promise<T>): Promise<T> {
		if (this.isOpen()) {
			throw new Error('Circuit breaker is open')
		}

		try {
			const result = await fn()
			this.failures = 0
			return result
		} catch (error) {
			this.failures++
			this.lastFailure = Date.now()
			throw error
		}
	}

	private isOpen(): boolean {
		if (this.failures < this.threshold) return false
		return Date.now() - this.lastFailure < this.resetMs
	}
}
```

---

---
url: https://chrisrecalis.github.io/kafkats/client/advanced/cluster.md
---
# Cluster API

The Cluster class manages broker connections and metadata discovery. It's used internally by Producer and Consumer but can be accessed directly for advanced use cases.

## Accessing the Cluster

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})

// Access the internal cluster
const cluster = client.cluster
```

## Cluster Metadata

Get information about the cluster:

```typescript
// Fetch fresh metadata
const metadata = await cluster.fetchMetadata()

console.log({
	brokers: metadata.brokers,
	topics: metadata.topics,
})
```

### Metadata Structure

```typescript
interface ClusterMetadata {
	brokers: BrokerInfo[]
	topics: TopicMetadata[]
	controllerId: number
}

interface BrokerInfo {
	nodeId: number
	host: string
	port: number
	rack?: string
}

interface TopicMetadata {
	name: string
	partitions: PartitionMetadata[]
	isInternal: boolean
}

interface PartitionMetadata {
	partitionIndex: number
	leader: number
	replicas: number[]
	isr: number[] // In-sync replicas
}
```

## Getting Brokers

```typescript
// Get broker for a specific partition
const broker = await cluster.getBrokerForPartition('my-topic', 0)

// Get the controller broker
const controller = await cluster.getController()

// Get group coordinator
const coordinator = await cluster.findGroupCoordinator('my-group')
```

## Topic Management

### List Topics

```typescript
const metadata = await cluster.fetchMetadata()
const topicNames = metadata.topics.map(t => t.name)
```

### Get Partition Count

```typescript
const metadata = await cluster.fetchMetadata({ topics: ['my-topic'] })
const topic = metadata.topics.find(t => t.name === 'my-topic')
const partitionCount = topic?.partitions.length ?? 0
```

### Get Partition Leaders

```typescript
const metadata = await cluster.fetchMetadata({ topics: ['my-topic'] })
const topic = metadata.topics.find(t => t.name === 'my-topic')

for (const partition of topic?.partitions ?? []) {
	console.log(`Partition ${partition.partitionIndex}: leader=${partition.leader}`)
}
```

## Metadata Refresh

Metadata is cached and refreshed automatically. Force a refresh:

```typescript
// Force metadata refresh
await cluster.refreshMetadata()

// Refresh metadata for specific topics
await cluster.refreshMetadata({ topics: ['topic-a', 'topic-b'] })
```

## Connection Management

The cluster manages a pool of connections to brokers:

```typescript
// Connections are created on-demand and reused
const broker = await cluster.getBrokerForPartition('my-topic', 0)
// broker has an active connection
```

## Advanced Usage

### Direct Broker Access

For low-level operations:

```typescript
const broker = await cluster.getBrokerForPartition('my-topic', 0)

// Use broker APIs directly
const response = await broker.fetch({
	topics: [
		{
			topic: 'my-topic',
			partitions: [{ partition: 0, fetchOffset: 0n }],
		},
	],
})
```

### Custom Metadata Handling

```typescript
// Listen for metadata updates
cluster.on('metadataUpdate', metadata => {
	console.log('Metadata updated:', metadata)
})
```

## Error Handling

```typescript
import { BrokerNotAvailableError, LeaderNotAvailableError } from '@kafkats/client'

try {
	const broker = await cluster.getBrokerForPartition('my-topic', 0)
} catch (error) {
	if (error instanceof LeaderNotAvailableError) {
		// Wait for leader election
		await delay(1000)
		// Retry
	}
}
```

## Best Practices

1. **Let kafkats manage connections** - Use Producer/Consumer APIs when possible
2. **Avoid caching metadata** - It can become stale
3. **Handle retriable errors** - Leader changes are normal
4. **Close properly** - Cluster is closed when client is closed

---

---
url: https://chrisrecalis.github.io/kafkats/client/advanced/broker.md
---
# Broker API

The Broker class provides typed protocol operations on a single Kafka broker connection. It's the lowest-level API for direct Kafka protocol access.

## Getting a Broker

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})

// Get broker for a partition
const broker = await client.cluster.getBrokerForPartition('my-topic', 0)

// Get group coordinator
const coordinator = await client.cluster.findGroupCoordinator('my-group')
```

## Produce API

Low-level produce request:

```typescript
import { RecordBatch } from '@kafkats/client'

const response = await broker.produce({
	topics: [
		{
			topic: 'my-topic',
			partitions: [
				{
					partition: 0,
					records: recordBatch,
				},
			],
		},
	],
	acks: -1, // Wait for all replicas
	timeoutMs: 30000,
})

for (const topic of response.topics) {
	for (const partition of topic.partitions) {
		console.log(`Offset: ${partition.baseOffset}`)
	}
}
```

## Fetch API

Low-level fetch request:

```typescript
const response = await broker.fetch({
	topics: [
		{
			topic: 'my-topic',
			partitions: [
				{
					partition: 0,
					fetchOffset: 0n,
					maxBytes: 1048576,
				},
			],
		},
	],
	maxWaitMs: 5000,
	minBytes: 1,
	maxBytes: 10485760,
	isolationLevel: 0,
})

for (const topic of response.topics) {
	for (const partition of topic.partitions) {
		console.log(`High watermark: ${partition.highWatermark}`)
		// Process partition.records
	}
}
```

## Metadata API

```typescript
const metadata = await broker.metadata({
	topics: ['my-topic'],
	allowAutoTopicCreation: false,
})

for (const topic of metadata.topics) {
	console.log(`Topic: ${topic.name}, Partitions: ${topic.partitions.length}`)
}
```

## Offset APIs

### List Offsets

```typescript
const response = await broker.listOffsets({
	topics: [
		{
			topic: 'my-topic',
			partitions: [
				{
					partition: 0,
					timestamp: -1n, // Latest offset
				},
			],
		},
	],
})

// -1n = latest, -2n = earliest
```

### Commit Offsets

```typescript
await broker.offsetCommit({
	groupId: 'my-group',
	generationId: 1,
	memberId: 'member-id',
	topics: [
		{
			topic: 'my-topic',
			partitions: [
				{
					partition: 0,
					committedOffset: 100n,
				},
			],
		},
	],
})
```

### Fetch Committed Offsets

```typescript
const response = await broker.offsetFetch({
	groupId: 'my-group',
	topics: [
		{
			topic: 'my-topic',
			partitions: [0, 1, 2],
		},
	],
})
```

## Consumer Group APIs

### Find Coordinator

```typescript
const response = await broker.findCoordinator({
	key: 'my-group',
	keyType: 0, // 0 = group, 1 = transaction
})

console.log(`Coordinator: ${response.host}:${response.port}`)
```

### Join Group

```typescript
const response = await broker.joinGroup({
	groupId: 'my-group',
	sessionTimeoutMs: 30000,
	rebalanceTimeoutMs: 60000,
	memberId: '',
	protocolType: 'consumer',
	protocols: [
		{
			name: 'range',
			metadata: subscriptionMetadata,
		},
	],
})

console.log(`Member ID: ${response.memberId}`)
console.log(`Leader: ${response.leader}`)
```

### Sync Group

```typescript
const response = await broker.syncGroup({
	groupId: 'my-group',
	generationId: 1,
	memberId: 'member-id',
	assignments: [
		{
			memberId: 'member-id',
			assignment: assignmentData,
		},
	],
})
```

### Heartbeat

```typescript
await broker.heartbeat({
	groupId: 'my-group',
	generationId: 1,
	memberId: 'member-id',
})
```

### Leave Group

```typescript
await broker.leaveGroup({
	groupId: 'my-group',
	members: [
		{
			memberId: 'member-id',
		},
	],
})
```

## Transaction APIs

### Init Producer ID

```typescript
const response = await broker.initProducerId({
	transactionalId: 'my-txn',
	transactionTimeoutMs: 60000,
})

console.log(`Producer ID: ${response.producerId}`)
console.log(`Producer Epoch: ${response.producerEpoch}`)
```

### Add Partitions to Transaction

```typescript
await broker.addPartitionsToTxn({
	transactionalId: 'my-txn',
	producerId: 123n,
	producerEpoch: 0,
	topics: [
		{
			topic: 'my-topic',
			partitions: [0, 1],
		},
	],
})
```

### End Transaction

```typescript
await broker.endTxn({
	transactionalId: 'my-txn',
	producerId: 123n,
	producerEpoch: 0,
	committed: true, // or false to abort
})
```

## API Versions

Check supported API versions:

```typescript
const versions = await broker.apiVersions()

for (const api of versions.apiKeys) {
	console.log(`API ${api.apiKey}: versions ${api.minVersion}-${api.maxVersion}`)
}
```

## Error Handling

All broker operations can throw protocol errors:

```typescript
import { KafkaProtocolError } from '@kafkats/client'

try {
  await broker.produce({...})
} catch (error) {
  if (error instanceof KafkaProtocolError) {
    console.log('Error code:', error.code)
    console.log('Error message:', error.message)
  }
}
```

## Best Practices

1. **Use high-level APIs** - Producer/Consumer handle retries and metadata
2. **Check API versions** - Not all brokers support all APIs
3. **Handle errors** - Broker operations can fail for many reasons
4. **Don't cache brokers** - Leadership can change

---

---
url: https://chrisrecalis.github.io/kafkats/client/advanced/protocol.md
---
# Protocol Internals

kafkats implements the Kafka wire protocol directly in TypeScript. This page covers the protocol layer for advanced users who need low-level access.

## Binary Encoding

### Encoder

The Encoder class builds binary buffers:

```typescript
import { Encoder } from '@kafkats/client'

const encoder = new Encoder()
encoder.writeInt32(42)
encoder.writeString('hello')
encoder.writeBytes(Buffer.from([1, 2, 3]))

const buffer = encoder.toBuffer()
```

### Encoder Methods

| Method                  | Description                          |
| ----------------------- | ------------------------------------ |
| `writeInt8(n)`          | Write signed 8-bit integer           |
| `writeInt16(n)`         | Write signed 16-bit integer          |
| `writeInt32(n)`         | Write signed 32-bit integer          |
| `writeInt64(n)`         | Write signed 64-bit integer (bigint) |
| `writeUInt32(n)`        | Write unsigned 32-bit integer        |
| `writeVarInt(n)`        | Write variable-length integer        |
| `writeVarLong(n)`       | Write variable-length long (bigint)  |
| `writeString(s)`        | Write length-prefixed string         |
| `writeBytes(b)`         | Write length-prefixed bytes          |
| `writeCompactString(s)` | Write compact string (varint length) |
| `writeCompactBytes(b)`  | Write compact bytes (varint length)  |
| `writeArray(arr, fn)`   | Write array with encoder function    |

### Size Calculation

Pre-calculate buffer sizes for efficiency:

```typescript
const size = Encoder.sizeOfInt32() + Encoder.sizeOfString('hello') + Encoder.sizeOfBytes(data)

const encoder = new Encoder(size) // Pre-allocated
```

### Decoder

The Decoder class reads binary buffers:

```typescript
import { Decoder } from '@kafkats/client'

const decoder = new Decoder(buffer)
const num = decoder.readInt32()
const str = decoder.readString()
const bytes = decoder.readBytes()
```

### Decoder Methods

| Method                | Description                         |
| --------------------- | ----------------------------------- |
| `readInt8()`          | Read signed 8-bit integer           |
| `readInt16()`         | Read signed 16-bit integer          |
| `readInt32()`         | Read signed 32-bit integer          |
| `readInt64()`         | Read signed 64-bit integer (bigint) |
| `readUInt32()`        | Read unsigned 32-bit integer        |
| `readVarInt()`        | Read variable-length integer        |
| `readVarLong()`       | Read variable-length long (bigint)  |
| `readString()`        | Read length-prefixed string         |
| `readBytes()`         | Read length-prefixed bytes          |
| `readCompactString()` | Read compact string                 |
| `readCompactBytes()`  | Read compact bytes                  |
| `readArray(fn)`       | Read array with decoder function    |

## Record Batches

Kafka messages are grouped into record batches:

```typescript
import { RecordBatch, Record } from '@kafkats/client'

// Create a record
const record = Record.create({
	key: Buffer.from('key'),
	value: Buffer.from('value'),
	headers: { header: Buffer.from('value') },
	timestamp: Date.now(),
})

// Create a batch
const batch = RecordBatch.create({
	records: [record],
	compression: 0, // 0=none, 1=gzip, 2=snappy, 3=lz4, 4=zstd
})
```

### RecordBatch Structure

```typescript
interface RecordBatch {
	baseOffset: bigint
	batchLength: number
	partitionLeaderEpoch: number
	magic: number // Always 2 for current format
	crc: number
	attributes: number
	lastOffsetDelta: number
	baseTimestamp: bigint
	maxTimestamp: bigint
	producerId: bigint
	producerEpoch: number
	baseSequence: number
	records: Record[]
}
```

### Record Structure

```typescript
interface Record {
	length: number
	attributes: number
	timestampDelta: bigint
	offsetDelta: number
	key: Buffer | null
	value: Buffer
	headers: Array<{ key: string; value: Buffer }>
}
```

## Protocol Requests

Access raw request/response types:

```typescript
import { requests, responses } from '@kafkats/client'

// Request types
type ProduceRequest = requests.ProduceRequest
type FetchRequest = requests.FetchRequest

// Response types
type ProduceResponse = responses.ProduceResponse
type FetchResponse = responses.FetchResponse
```

## API Keys

Kafka API identifiers:

```typescript
import { ApiKeys } from '@kafkats/client'

ApiKeys.Produce // 0
ApiKeys.Fetch // 1
ApiKeys.ListOffsets // 2
ApiKeys.Metadata // 3
// ... etc
```

## Error Codes

Kafka protocol error codes:

```typescript
import { ErrorCode } from '@kafkats/client'

ErrorCode.None // 0
ErrorCode.UnknownTopicOrPartition // 3
ErrorCode.LeaderNotAvailable // 5
ErrorCode.NotLeaderForPartition // 6
// ... etc
```

## Request Header

All requests include a header:

```typescript
interface RequestHeader {
	apiKey: number
	apiVersion: number
	correlationId: number
	clientId: string
}
```

## Response Header

All responses include a header:

```typescript
interface ResponseHeader {
	correlationId: number
}
```

## Compression

Record batches can be compressed:

```typescript
import { compress, decompress, CompressionType } from '@kafkats/client'

// Compress
const compressed = await compress(CompressionType.Snappy, uncompressedBuffer)

// Decompress
const decompressed = await decompress(CompressionType.Snappy, compressedBuffer)
```

### Compression Types

| Value | Name   |
| ----- | ------ |
| 0     | None   |
| 1     | GZIP   |
| 2     | Snappy |
| 3     | LZ4    |
| 4     | ZSTD   |

## Variable-Length Encoding

Kafka uses variable-length integers for efficiency:

```typescript
// VarInt (signed, zig-zag encoded)
const encoded = Encoder.sizeOfVarInt(value)

// VarLong (64-bit, zig-zag encoded)
const encoded = Encoder.sizeOfVarLong(value)
```

## CRC32C

Kafka uses CRC32C for checksums:

```typescript
import { crc32c, verifyCrc32c } from '@kafkats/client'

const checksum = crc32c(buffer)
const isValid = verifyCrc32c(buffer, expectedChecksum)
```

## Best Practices

1. **Use high-level APIs** - Protocol details are abstracted
2. **Pre-calculate sizes** - For better performance
3. **Handle all error codes** - Many operations can fail
4. **Check API versions** - Protocol evolves over time

---

---
url: https://chrisrecalis.github.io/kafkats/flow.md
---
# @kafkats/flow

Kafka Streams-like flow APIs for building stream processing applications in TypeScript.

## Features

* **Kafka Streams DSL** - Familiar APIs: KStream, KTable, windowing, joins
* **Exactly-Once Semantics** - Transactional processing with batch commits
* **Type-Safe** - Full TypeScript support with strong typing
* **Pluggable State** - In-memory and LMDB state stores
* **Windowing** - Time, session, and sliding windows
* **Joins** - Stream-stream and stream-table joins
* **Testing** - Built-in test utilities

## Installation

```bash
pnpm add @kafkats/flow
```

## Quick Example

```typescript
import { flow, topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

interface ClickEvent {
	userId: string
	page: string
}

interface ClickCount {
	userId: string
	count: number
}

// Define topics
const clicks = topic('clicks', {
	key: string(),
	value: json<ClickEvent>(),
})

const counts = topic('click-counts', {
	key: string(),
	value: json<ClickCount>(),
})

// Create app
const app = flow({
	applicationId: 'click-counter',
	client: { clientId: 'click-counter', brokers: ['localhost:9092'] },
})

// Build topology
app.stream(clicks)
	.groupByKey()
	.count()
	.toStream()
	.mapValues((count, key) => ({ userId: key, count }))
	.to(counts)

// Start processing
await app.start()
```

## Core Concepts

### KStream

An unbounded stream of key-value records. Each record is an independent event.

```typescript
app.stream(inputTopic)
	.filter((key, value) => value.amount > 100)
	.mapValues(value => ({ ...value, processed: true }))
	.to(outputTopic)
```

### KTable

A changelog stream representing a table. Each key has a latest value.

```typescript
const usersTable = app.table(usersTopic)
usersTable.mapValues(user => user.email)
```

### Windowing

Group records by time for aggregations:

```typescript
import { TimeWindows } from '@kafkats/flow'

app.stream(clicks).groupByKey().windowedBy(TimeWindows.of('5m')).count()
```

## Architecture

```
@kafkats/flow
├── flow()           # Create streaming application
├── topic()          # Define typed topics
├── KStream          # Unbounded record stream
├── KTable           # Changelog table
├── KGroupedStream   # Grouped stream for aggregations
├── Windowing        # Time, session, sliding windows
└── State Stores     # In-memory and persistent

@kafkats/client (codecs)
├── string()         # UTF-8 string codec
├── json<T>()        # JSON codec with types
├── buffer()         # Raw buffer codec
└── codec()          # Custom codec factory
```

## Next Steps

* [Getting Started](/flow/getting-started) - Setup and basic usage
* [KStream](/flow/streams) - Stream operations
* [KTable](/flow/tables) - Table operations
* [Windowing](/flow/windowing) - Time-based processing
* [State Stores](/flow/state-stores) - Stateful processing

---

---
url: https://chrisrecalis.github.io/kafkats/flow/getting-started.md
---
# Getting Started with @kafkats/flow

## Installation

```bash
pnpm add @kafkats/flow
```

## Creating a Flow Application

```typescript
import { flow, topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

const app = flow({
	applicationId: 'my-stream-app',
	client: {
		clientId: 'my-stream-app',
		brokers: ['localhost:9092'],
	},
})
```

### Configuration Options

| Option                | Type                                | Default           | Description                                                                                   |
| --------------------- | ----------------------------------- | ----------------- | --------------------------------------------------------------------------------------------- |
| `applicationId`       | `string`                            | -                 | Required. Also used as the consumer group id                                                  |
| `client`              | `KafkaClient \| KafkaClientConfig`  | -                 | Required. Pass an existing client or a config object                                          |
| `numStreamThreads`    | `number`                            | `1`               | Number of parallel stream threads (each has its own producer/consumer)                        |
| `processingGuarantee` | `'at_least_once' \| 'exactly_once'` | `'at_least_once'` | Enables transactional processing when `'exactly_once'`                                        |
| `commitIntervalMs`    | `number`                            | `100`             | Transaction commit interval in milliseconds (only applies to `exactly_once`)                  |
| `stateDir`            | `string`                            | -                 | State directory (used by some store providers)                                                |
| `stateStoreProvider`  | `StateStoreProvider`                | in-memory         | State store backend (in-memory by default)                                                    |
| `changelog`           | `object`                            | -                 | Changelog topic settings: `replicationFactor`, `topicConfigs`, `autoCreate`                   |
| `consumer`            | `Omit<ConsumerConfig, 'groupId'>`   | -                 | Consumer overrides (Flow sets `groupId` to `applicationId`)                                   |
| `producer`            | `ProducerConfig`                    | -                 | Producer overrides                                                                            |
| `runEach`             | `RunEachOptions`                    | -                 | Consumer run-loop options (e.g. `partitionConcurrency`, `autoCommitIntervalMs`, `assignment`) |

#### Client Config (KafkaClientConfig)

If you pass a config object as `client`, it uses the same options as `new KafkaClient({...})` in `@kafkats/client`:

| Option     | Type         | Notes                                               |
| ---------- | ------------ | --------------------------------------------------- |
| `brokers`  | `string[]`   | Required                                            |
| `clientId` | `string`     | Optional (Flow may set a default)                   |
| `tls`      | `TlsConfig`  | Omit for plaintext; use `{ enabled: true }` for TLS |
| `sasl`     | `SaslConfig` | SASL authentication                                 |

## Processing Guarantees

Flow supports two processing guarantees:

### At-Least-Once (Default)

Messages are processed at least once. In case of failures, some messages may be reprocessed.

```typescript
const app = flow({
	applicationId: 'my-app',
	client: { brokers: ['localhost:9092'] },
	processingGuarantee: 'at_least_once', // default
})
```

Consumer offsets are committed periodically (controlled by `autoCommitIntervalMs` in `runEach` options).

### Exactly-Once

Messages are processed exactly once using Kafka transactions. Output messages and consumer offset commits are atomic.

```typescript
const app = flow({
	applicationId: 'my-app',
	client: { brokers: ['localhost:9092'] },
	processingGuarantee: 'exactly_once',
	commitIntervalMs: 100, // optional, default 100ms
})
```

#### How Exactly-Once Works

Flow implements exactly-once semantics using batch-based transaction commits, similar to Kafka Streams:

1. **Batch Processing** - Multiple input messages are processed within a single transaction batch
2. **Periodic Commits** - Transactions commit at regular intervals (controlled by `commitIntervalMs`)
3. **Atomic Commits** - Each commit atomically writes output messages and commits consumer offsets
4. **Replica-Safe IDs** - Flow gives each application instance an internal process UUID and derives each
   producer's transactional ID as `<applicationId>-<processId>-w<thread>`

The process UUID is generated when the `FlowApp` is created, so replicas can safely share the normal
`applicationId` and `clientId` configuration. It is not currently persisted: after a process restart, Flow uses a
new UUID. Exactly-once correctness is preserved, but a transaction left open by the previous process may remain
open until its transaction timeout expires.

If `producer.transactionalId` is configured explicitly, the caller is responsible for making it unique among
simultaneously running application instances. Flow appends a worker suffix when `numStreamThreads` is greater
than one.

Transactions are also committed:

* When the application shuts down via `close()`
* When a consumer rebalance occurs (partitions are revoked)

#### Commit Interval Tuning

The `commitIntervalMs` setting controls the trade-off between latency and throughput:

| Value            | Latency | Throughput | Use Case                 |
| ---------------- | ------- | ---------- | ------------------------ |
| Lower (50-100ms) | Lower   | Lower      | Real-time processing     |
| Higher (500ms+)  | Higher  | Higher     | Batch-oriented workloads |

::: tip
The default of 100ms provides a good balance for most use cases. Kafka Streams uses a default of 30 seconds, but Flow uses a lower default for more responsive processing.
:::

#### Consumer Configuration

When using exactly-once, downstream consumers should use `read_committed` isolation to only see committed messages:

```typescript
const consumer = client.consumer({
	groupId: 'downstream-consumer',
	isolationLevel: 'read_committed',
})
```

## Defining Topics

Define typed topics for input and output:

```typescript
import { topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

interface UserEvent {
	userId: string
	action: string
	timestamp: number
}

const events = topic('user-events', {
	key: string(),
	value: json<UserEvent>(),
})
```

## Building a Topology

### Stream Processing

```typescript
// Create a stream from a topic
app.stream(events)
	.filter((key, value) => value.action === 'purchase')
	.mapValues(value => ({ ...value, processed: true }))
	.to(outputTopic)
```

### Table Processing

```typescript
// Create a table (changelog) from a topic
const usersTable = app.table(usersTopic)

// Transform table values
usersTable
	.mapValues(user => ({ name: user.name, email: user.email }))
	.toStream()
	.to(userProfilesTopic)
```

### Aggregations

```typescript
app.stream(events).groupByKey().count().toStream().to(countsTopic)
```

## Starting and Stopping

```typescript
// Start the application
await app.start()

// Check state
console.log(app.state()) // 'RUNNING'

// Stop gracefully
await app.close()
```

### Application States

| State         | Description                      |
| ------------- | -------------------------------- |
| `CREATED`     | Application created, not started |
| `RUNNING`     | Processing messages              |
| `REBALANCING` | Consumer group rebalancing       |
| `ERROR`       | Fatal error occurred             |
| `STOPPED`     | Gracefully stopped               |

## Complete Example

```typescript
import { flow, topic, TimeWindows } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

interface PageView {
	userId: string
	page: string
	timestamp: number
}

interface PageViewCount {
	page: string
	count: number
	windowStart: number
	windowEnd: number
}

// Define topics
const pageViews = topic('page-views', {
	key: string(),
	value: json<PageView>(),
})

const pageViewCounts = topic('page-view-counts', {
	key: string(),
	value: json<PageViewCount>(),
})

// Create app
const app = flow({
	applicationId: 'page-view-counter',
	client: { clientId: 'page-view-counter', brokers: ['localhost:9092'] },
})

// Build topology
app.stream(pageViews)
	// Rekey by page
	.selectKey((_, value) => value.page)
	// Group and window
	.groupByKey()
	.windowedBy(TimeWindows.of('1h'))
	.count()
	// Transform output
	.toStream()
	.map((windowedKey, count) => ({
		key: windowedKey.key,
		value: {
			page: windowedKey.key,
			count,
			windowStart: windowedKey.window.start,
			windowEnd: windowedKey.window.end,
		},
	}))
	.to(pageViewCounts)

// Handle shutdown
process.on('SIGTERM', async () => {
	await app.close()
})

// Start
await app.start()
console.log('Stream processing started')
```

## Next Steps

* [KStream Operations](/flow/streams) - Stream transformations
* [KTable Operations](/flow/tables) - Table transformations
* [Aggregations](/flow/aggregations) - Counting and reducing
* [Windowing](/flow/windowing) - Time-based processing

---

---
url: https://chrisrecalis.github.io/kafkats/flow/streams.md
---
# KStream

A KStream represents an unbounded stream of key-value records. Each record is an independent event.

## Creating a Stream

```typescript
import { flow, topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
})

// From a topic
const stream = app.stream(myTopic)

// With explicit types
const stream = app.stream<string, MyEvent>(myTopic)
```

## Transformation Operations

### map

Transform both key and value:

```typescript
stream.map((key, value) => ({
	key: value.userId,
	value: { ...value, processed: true },
}))
```

### mapValues

Transform only the value (preserves key):

```typescript
stream.mapValues(value => ({
	...value,
	timestamp: Date.now(),
}))
```

### selectKey

Change the key:

```typescript
stream.selectKey((key, value) => value.userId)
```

### filter

Keep only matching records:

```typescript
stream.filter((key, value) => value.amount > 100)
```

### filterNot

Remove matching records:

```typescript
stream.filterNot((key, value) => value.deleted)
```

### flatMap

Emit multiple records per input:

```typescript
stream.flatMap((key, value) => [
	{ key: `${key}-1`, value: value.part1 },
	{ key: `${key}-2`, value: value.part2 },
])
```

### flatMapValues

Emit multiple values per input:

```typescript
stream.flatMapValues(value => value.items)
```

## Side Effects

### peek

Perform side effects without modifying the stream:

```typescript
stream.peek((key, value) => {
	console.log(`Processing: ${key}`)
	metrics.increment('processed')
})
```

## Branching

### branch

Split stream into multiple branches:

```typescript
const [highValue, lowValue] = stream.branch(
	(key, value) => value.amount > 1000,
	(key, value) => value.amount <= 1000
)

highValue.to(highValueTopic)
lowValue.to(lowValueTopic)
```

## Merging

### merge

Combine multiple streams:

```typescript
const merged = stream1.merge(stream2)

// Or merge multiple
const merged = stream1.merge(stream2, stream3, stream4)
```

## Output

### to

Write to a topic (terminal operation):

```typescript
stream.to(outputTopic)

// With options
stream.to(outputTopic, {
	partitioner: (key, _value, partitionCount) => {
		if (key === null) return 0
		return String(key).length % partitionCount
	},
})
```

### through

Write to topic and continue processing:

```typescript
stream
	.through(intermediateTopic)
	.mapValues(value => transform(value))
	.to(finalTopic)
```

## Grouping

### groupByKey

Group by existing key:

```typescript
const grouped = stream.groupByKey()
// Returns KGroupedStream
```

### groupBy

Group by a new key:

```typescript
const grouped = stream.groupBy((key, value) => value.category)
```

## Conversion

### toTable

Convert stream to table:

```typescript
const table = stream.toTable()
```

## Example: Event Processing Pipeline

```typescript
interface RawEvent {
	type: string
	userId: string
	data: unknown
	timestamp: number
}

interface ProcessedEvent {
	type: string
	userId: string
	data: unknown
	processedAt: number
	source: string
}

app.stream<string, RawEvent>(rawEvents)
	// Filter valid events
	.filter((_, event) => event.type !== 'heartbeat')

	// Enrich
	.mapValues(
		event =>
			({
				...event,
				processedAt: Date.now(),
				source: 'stream-processor',
			}) as ProcessedEvent
	)

	// Log
	.peek((key, event) => {
		logger.debug(`Processing event ${event.type} for user ${event.userId}`)
	})

	// Route by type
	.branch(
		(_, e) => e.type === 'purchase',
		(_, e) => e.type === 'pageview',
		() => true // default
	)
	.forEach((branch, index) => {
		const topics = [purchasesTopic, pageviewsTopic, otherEventsTopic]
		branch.to(topics[index])
	})
```

## Type Parameters

KStream operations preserve and transform types:

```typescript
// KStream<string, UserEvent>
const stream = app.stream(userEvents)

// KStream<string, { count: number }> - value type changed
const mapped = stream.mapValues(e => ({ count: 1 }))

// KStream<string, { count: number }> - key type changed to userId
const rekeyed = mapped.selectKey((_, v) => v.userId)
```

---

---
url: https://chrisrecalis.github.io/kafkats/flow/tables.md
---
# KTable

A KTable represents a changelog stream where each key has a latest value. It's like a continuously-updated database table.

## Creating a Table

```typescript
import { flow, topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
})

// From a topic
const usersTable = app.table(usersTopic)

// Global table (fully replicated to all instances)
const configTable = app.globalTable(configTopic)
```

## Table vs Stream

| Aspect     | KStream        | KTable               |
| ---------- | -------------- | -------------------- |
| Semantics  | Event log      | Latest value per key |
| Null value | Regular record | Delete (tombstone)   |
| Use case   | Events, logs   | State, lookups       |

## Transformation Operations

### mapValues

Transform table values:

```typescript
usersTable.mapValues(user => ({
	name: user.name,
	email: user.email,
}))
```

### filter

Keep only matching entries:

```typescript
usersTable.filter((key, user) => user.isActive)
```

### filterNot

Remove matching entries:

```typescript
usersTable.filterNot((key, user) => user.deleted)
```

## Conversion

### toStream

Convert table to stream:

```typescript
const stream = usersTable.toStream()

// Now you can use stream operations
stream.mapValues(user => ({ event: 'user_updated', user })).to(userEventsTopic)
```

## Grouping

### groupBy

Group table by a new key:

```typescript
const byCountry = usersTable.groupBy((key, user) => user.country)
// Returns KGroupedTable
```

## Joins

### join (inner join)

Join with another table:

```typescript
const ordersWithUsers = ordersTable.join(usersTable, (order, user) => ({
	orderId: order.id,
	userName: user.name,
	total: order.total,
}))
```

### leftJoin

Left join - keep all left records:

```typescript
const ordersWithOptionalUser = ordersTable.leftJoin(usersTable, (order, user) => ({
	orderId: order.id,
	userName: user?.name ?? 'Unknown',
	total: order.total,
}))
```

### outerJoin

Outer join - keep all records from both:

```typescript
const merged = table1.outerJoin(table2, (left, right) => ({
	fromLeft: left?.value,
	fromRight: right?.value,
}))
```

## Global Tables

Global tables are fully replicated to all application instances, useful for small lookup tables:

```typescript
// Define a config topic
const configTopic = topic('app-config', {
	key: string(),
	value: json<ConfigValue>(),
})

// Create global table
const configTable = app.globalTable(configTopic)

// Use in joins - available on all partitions
app.stream(eventsTopic)
	.join(configTable, (event, config) => ({
		...event,
		settings: config,
	}))
	.to(enrichedEventsTopic)
```

::: tip When to Use Global Tables

* Small, slowly-changing data (config, reference data)
* Data needed for every partition
* Lookup tables for enrichment
  :::

## Example: User Enrichment

```typescript
interface User {
	id: string
	name: string
	email: string
	tier: 'free' | 'premium'
}

interface Order {
	orderId: string
	userId: string
	items: string[]
	total: number
}

interface EnrichedOrder {
	orderId: string
	userId: string
	userName: string
	userTier: string
	items: string[]
	total: number
	discount: number
}

// Users table from compacted topic
const usersTable = app.table<string, User>(usersTopic)

// Orders stream
app.stream<string, Order>(ordersTopic)
	// Rekey by userId for join
	.selectKey((_, order) => order.userId)
	// Join with users
	.join(
		usersTable,
		(order, user) =>
			({
				orderId: order.orderId,
				userId: order.userId,
				userName: user.name,
				userTier: user.tier,
				items: order.items,
				total: order.total,
				discount: user.tier === 'premium' ? order.total * 0.1 : 0,
			}) as EnrichedOrder
	)
	.to(enrichedOrdersTopic)
```

## Table State

Tables maintain state internally. Access the underlying store:

```typescript
// Materialize table to a named store
const users = app.table(usersTopic, {
	materialized: { storeName: 'users-store' },
})

// Query the store (after app is running)
const store = app.getStore<string, User>('users-store')
const user = await store.get('user-123')
```

## Tombstones (Deletions)

In tables, a null value means "delete this key":

```typescript
// Delete a user by sending null
await producer.send(usersTopic, [{ key: 'user-123', value: null }])

// The key is removed from the table
```

---

---
url: https://chrisrecalis.github.io/kafkats/flow/operations.md
---
# Stream Operations

Complete reference of all KStream and KTable operations.

## Options Reference

Many operations accept an `options` object. These are the common ones:

### Consumed (for `app.stream(...)`, `app.table(...)`)

| Option        | Type                               | Description                                |
| ------------- | ---------------------------------- | ------------------------------------------ |
| `key`         | `Codec<K>`                         | Decode keys                                |
| `value`       | `Codec<V>`                         | Decode values                              |
| `offsetReset` | `'earliest' \| 'latest' \| 'none'` | What to do when no committed offset exists |

### Produced (for `to(...)`, `through(...)`)

| Option | Type | Description |
| ------------- | ---------- | -------------------------------------------------- | --------------------------------------- |
| `key` | `Codec<K>` | Encode keys |
| `value` | `Codec<V>` | Encode values |
| `partitioner` | `(key: K   | null, value: V, partitionCount: number) => number` | Choose a partition for produced records |

### Grouped (for `groupBy(...)`, `groupByKey(...)`)

| Option  | Type       | Description                               |
| ------- | ---------- | ----------------------------------------- |
| `key`   | `Codec<K>` | Key codec for grouping / repartitioning   |
| `value` | `Codec<V>` | Value codec for grouping / repartitioning |

### Materialized (for `toTable(...)` and stateful ops)

| Option      | Type       | Description                                  |
| ----------- | ---------- | -------------------------------------------- |
| `storeName` | `string`   | Store name (and changelog topic name prefix) |
| `key`       | `Codec<K>` | Key codec for the state store                |
| `value`     | `Codec<V>` | Value codec for the state store              |

### Joined (for `join(...)`, `leftJoin(...)`, `outerJoin(...)`)

| Option       | Type                                              | Description                                    |
| ------------ | ------------------------------------------------- | ---------------------------------------------- |
| `key`        | `Codec<K>`                                        | Key codec                                      |
| `value`      | `Codec<V1>`                                       | Left-side value codec                          |
| `otherValue` | `Codec<V2>`                                       | Right-side value codec                         |
| `within`     | `TimeWindows \| SessionWindows \| SlidingWindows` | Join window (required for stream-stream joins) |

## Stateless Operations

These operations don't require state storage:

### map

Transform key and value:

```typescript
stream.map((key, value) => ({
	key: newKey,
	value: newValue,
}))
```

### mapValues

Transform only value:

```typescript
stream.mapValues(value => transformedValue)

// With key access
stream.mapValues((value, key) => ({ ...value, originalKey: key }))
```

### selectKey

Change the key:

```typescript
stream.selectKey((key, value) => value.userId)
```

### filter

Keep matching records:

```typescript
stream.filter((key, value) => condition)
```

### filterNot

Remove matching records:

```typescript
stream.filterNot((key, value) => condition)
```

### flatMap

One-to-many transformation:

```typescript
stream.flatMap((key, value) => [
	{ key: k1, value: v1 },
	{ key: k2, value: v2 },
])
```

### flatMapValues

One-to-many value transformation:

```typescript
stream.flatMapValues(value => [v1, v2, v3])
```

### peek

Side effects without transformation:

```typescript
stream.peek((key, value) => {
	console.log(key, value)
})
```

## Grouping Operations

### groupByKey

Group by existing key:

```typescript
const grouped = stream.groupByKey()
```

### groupBy

Group by computed key:

```typescript
const grouped = stream.groupBy((key, value) => value.category)

// With options
const grouped = stream.groupBy((key, value) => value.category, { key: string() })
```

## Aggregation Operations

Available on KGroupedStream and KGroupedTable:

### count

Count records per key:

```typescript
grouped.count()
// Returns KTable<K, number>
```

### reduce

Reduce to single value:

```typescript
grouped.reduce((agg, value) => agg + value.amount)
```

### aggregate

Custom aggregation:

```typescript
grouped.aggregate(
	() => ({ sum: 0, count: 0 }), // initializer
	(key, value, agg) => ({
		// aggregator
		sum: agg.sum + value.amount,
		count: agg.count + 1,
	})
)
```

## Windowed Operations

### windowedBy

Apply time window to grouped stream:

```typescript
import { TimeWindows } from '@kafkats/flow'

grouped.windowedBy(TimeWindows.of('5m'))
// Returns WindowedKGroupedStream
```

### sessionWindowedBy

Apply session window:

```typescript
import { SessionWindows } from '@kafkats/flow'

grouped.sessionWindowedBy(SessionWindows.withInactivityGap('30m'))
```

## Join Operations

### join (inner)

Inner join - both sides must have matching key:

```typescript
import { TimeWindows } from '@kafkats/flow'

// Stream-stream join
stream1.join(stream2, (v1, v2) => combined, {
	within: TimeWindows.of('5m'),
})

// Stream-table join
stream.join(table, (streamValue, tableValue) => combined)

// Table-table join
table1.join(table2, (v1, v2) => combined)
```

### leftJoin

Left join - keep all left records:

```typescript
stream.leftJoin(table, (streamValue, tableValue) => ({
	...streamValue,
	extra: tableValue?.field ?? 'default',
}))
```

### outerJoin

Outer join - keep all records from both:

```typescript
stream1.outerJoin(
	stream2,
	(v1, v2) => ({
		left: v1,
		right: v2,
	}),
	{ within: TimeWindows.of('5m') }
)
```

## Branching Operations

### branch

Split into multiple streams:

```typescript
const [premium, standard] = stream.branch(
	(key, value) => value.tier === 'premium',
	(key, value) => value.tier === 'standard'
)
```

### merge

Combine streams:

```typescript
const merged = stream1.merge(stream2, stream3)
```

## Output Operations

### to

Terminal - write to topic:

```typescript
stream.to(outputTopic)

// With options
const customPartitioner = (key, _value, partitionCount) => {
	if (key === null) return 0
	return String(key).length % partitionCount
}

stream.to(outputTopic, { partitioner: customPartitioner })
```

### through

Write and continue processing:

```typescript
stream
  .through(intermediateTopic)
  .filter(...)
  .to(finalTopic)
```

## Conversion Operations

### toTable

Convert stream to table:

```typescript
const table = stream.toTable()
```

### toStream

Convert table to stream:

```typescript
const stream = table.toStream()
```

## Chaining Operations

Operations can be chained fluently:

```typescript
app.stream(inputTopic)
	.filter((_, v) => v.valid)
	.mapValues(v => transform(v))
	.selectKey((_, v) => v.userId)
	.groupByKey()
	.windowedBy(TimeWindows.of('1h'))
	.count()
	.toStream()
	.map((windowedKey, count) => ({
		key: windowedKey.key,
		value: { userId: windowedKey.key, count, window: windowedKey.window },
	}))
	.to(outputTopic)
```

## Operation Categories

| Category    | Operations                         | State Required |
| ----------- | ---------------------------------- | -------------- |
| Transform   | map, mapValues, selectKey, flatMap | No             |
| Filter      | filter, filterNot                  | No             |
| Side Effect | peek                               | No             |
| Group       | groupByKey, groupBy                | No             |
| Aggregate   | count, reduce, aggregate           | Yes            |
| Window      | windowedBy, sessionWindowedBy      | Yes            |
| Join        | join, leftJoin, outerJoin          | Yes            |
| Branch      | branch, merge                      | No             |
| Output      | to, through                        | No             |
| Convert     | toTable, toStream                  | Maybe          |

---

---
url: https://chrisrecalis.github.io/kafkats/flow/windowing.md
---
# Windowing

Windowing groups stream records by time for aggregations. kafkats/flow supports three window types.

## Time Windows

Fixed-size, non-overlapping time buckets:

```typescript
import { flow, topic, TimeWindows } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

// 5-minute tumbling windows
app.stream(clicks).groupByKey().windowedBy(TimeWindows.of('5m')).count()
```

### Window Duration Syntax

| Format    | Duration           |
| --------- | ------------------ |
| `'100ms'` | 100 milliseconds   |
| `'5s'`    | 5 seconds          |
| `'5m'`    | 5 minutes          |
| `'1h'`    | 1 hour             |
| `'1d'`    | 1 day              |
| `300000`  | 300000 ms (number) |

### Hopping Windows

Overlapping windows with custom advance:

```typescript
// 5-minute windows, advancing every 1 minute
TimeWindows.of('5m').advanceBy('1m')
```

```
Window 1: [0:00 - 0:05)
Window 2: [0:01 - 0:06)
Window 3: [0:02 - 0:07)
...
```

## Session Windows

Dynamic windows based on activity gaps:

```typescript
import { SessionWindows } from '@kafkats/flow'

// Session closes after 30 minutes of inactivity
app.stream(userActivity).groupByKey().windowedBy(SessionWindows.withInactivityGap('30m')).count()
```

Sessions:

* Start with first event for a key
* Extend with each new event
* Close after inactivity gap elapses

## Sliding Windows

Continuous, overlapping windows:

```typescript
import { SlidingWindows } from '@kafkats/flow'

// Look back 5 minutes from each event
app.stream(events).groupByKey().windowedBy(SlidingWindows.of('5m')).count()
```

## Window Operations

Once windowed, use aggregation operations:

```typescript
const windowed = stream.groupByKey().windowedBy(TimeWindows.of('1h'))

// Count per window
windowed.count()

// Sum per window
windowed.reduce((sum, value) => sum + value.amount)

// Custom aggregation
windowed.aggregate(
	() => ({ count: 0, total: 0 }),
	(key, value, agg) => ({
		count: agg.count + 1,
		total: agg.total + value.amount,
	})
)
```

## Windowed Keys

Windowed aggregations produce `Windowed<K>` keys:

```typescript
interface Windowed<K> {
	key: K
	window: {
		start: number // Window start timestamp (ms)
		end: number // Window end timestamp (ms)
	}
}
```

Access window boundaries in transformations:

```typescript
stream
	.groupByKey()
	.windowedBy(TimeWindows.of('5m'))
	.count()
	.toStream()
	.map((windowedKey, count) => ({
		key: windowedKey.key,
		value: {
			key: windowedKey.key,
			windowStart: new Date(windowedKey.window.start),
			windowEnd: new Date(windowedKey.window.end),
			count,
		},
	}))
```

## Grace Periods

Handle late-arriving data:

```typescript
// Accept events up to 10 minutes late
TimeWindows.of('5m').gracePeriod('10m')
```

Without grace period, late events are dropped. With grace period:

* Window remains open longer
* Late events are included in aggregation
* Higher memory usage

## Window Comparison

| Type            | Size    | Overlap    | Use Case                         |
| --------------- | ------- | ---------- | -------------------------------- |
| Time (Tumbling) | Fixed   | No         | Regular metrics, hourly counts   |
| Time (Hopping)  | Fixed   | Yes        | Smoothed averages, sliding stats |
| Session         | Dynamic | No         | User sessions, activity tracking |
| Sliding         | Fixed   | Continuous | Event-relative windows           |

## Example: Hourly Metrics

```typescript
interface MetricEvent {
	metricName: string
	value: number
	timestamp: number
}

interface HourlyMetric {
	metricName: string
	windowStart: Date
	windowEnd: Date
	sum: number
	count: number
	average: number
}

app.stream<string, MetricEvent>(metricsTopic)
	.groupByKey()
	.windowedBy(TimeWindows.of('1h'))
	.aggregate(
		() => ({ sum: 0, count: 0 }),
		(key, value, agg) => ({
			sum: agg.sum + value.value,
			count: agg.count + 1,
		})
	)
	.toStream()
	.map((windowedKey, agg) => ({
		key: windowedKey.key,
		value: {
			metricName: windowedKey.key,
			windowStart: new Date(windowedKey.window.start),
			windowEnd: new Date(windowedKey.window.end),
			sum: agg.sum,
			count: agg.count,
			average: agg.sum / agg.count,
		} as HourlyMetric,
	}))
	.to(hourlyMetricsTopic)
```

## Example: User Sessions

```typescript
interface UserAction {
	userId: string
	action: string
	timestamp: number
}

interface SessionSummary {
	userId: string
	sessionStart: Date
	sessionEnd: Date
	actionCount: number
	actions: string[]
}

app.stream<string, UserAction>(actionsTopic)
	.groupByKey()
	.windowedBy(SessionWindows.withInactivityGap('30m'))
	.aggregate(
		() => ({ actions: [] as string[] }),
		(key, action, agg) => ({
			actions: [...agg.actions, action.action],
		})
	)
	.toStream()
	.map((windowedKey, agg) => ({
		key: windowedKey.key,
		value: {
			userId: windowedKey.key,
			sessionStart: new Date(windowedKey.window.start),
			sessionEnd: new Date(windowedKey.window.end),
			actionCount: agg.actions.length,
			actions: agg.actions,
		} as SessionSummary,
	}))
	.to(sessionsTopic)
```

## State Management

Windowed operations require state stores. Configure via `stateStoreProvider`:

```typescript
import { inMemory } from '@kafkats/flow'
import { lmdb } from '@kafkats/flow-state-lmdb'

// In-memory (default)
const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
	stateStoreProvider: inMemory(),
})

// Persistent (LMDB)
const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
	stateStoreProvider: lmdb({ stateDir: './state' }),
})
```

---

---
url: https://chrisrecalis.github.io/kafkats/flow/aggregations.md
---
# Aggregations

Aggregations combine multiple records into summary results. They're available on grouped streams and tables.

## Grouping First

Before aggregating, group records by key:

```typescript
// Group by existing key
const grouped = stream.groupByKey()

// Group by computed key
const grouped = stream.groupBy((key, value) => value.category)
```

## Count

Count records per key:

```typescript
const counts = stream.groupByKey().count()
// Returns KTable<K, number>
```

### Windowed Count

```typescript
import { TimeWindows } from '@kafkats/flow'

const hourlyCounts = stream.groupByKey().windowedBy(TimeWindows.of('1h')).count()
```

## Reduce

Combine values into one using a reducer:

```typescript
// Sum amounts per key
const totals = stream.groupByKey().reduce((sum, value) => sum + value.amount)
```

The reducer receives:

* Previous aggregated value (or first value)
* Current value
* Returns new aggregated value

## Aggregate

Custom aggregation with initializer:

```typescript
interface Stats {
	count: number
	sum: number
	min: number
	max: number
}

const stats = stream.groupByKey().aggregate<Stats>(
	// Initializer - creates empty aggregate
	() => ({ count: 0, sum: 0, min: Infinity, max: -Infinity }),

	// Aggregator - combines value into aggregate
	(key, value, agg) => ({
		count: agg.count + 1,
		sum: agg.sum + value.amount,
		min: Math.min(agg.min, value.amount),
		max: Math.max(agg.max, value.amount),
	})
)
```

## Table Aggregations

KGroupedTable has different aggregation semantics:

```typescript
const grouped = table.groupBy((key, value) => value.category)

// Must provide both adder and subtractor
const counts = grouped.aggregate(
	() => 0,
	(key, value, agg) => agg + 1, // add
	(key, value, agg) => agg - 1 // subtract (when key changes/deleted)
)
```

## Materialization

Store aggregation results in a named store:

```typescript
const counts = stream.groupByKey().count({
	materialized: {
		storeName: 'my-counts-store',
	},
})

// Query the store later
const store = app.getStore<string, number>('my-counts-store')
const count = await store.get('some-key')
```

## Converting Results

Aggregations return KTable. Convert to stream for output:

```typescript
stream.groupByKey().count().toStream().to(countsTopic)
```

## Example: Real-time Analytics

```typescript
interface PageView {
	page: string
	userId: string
	timestamp: number
	duration: number
}

interface PageStats {
	page: string
	views: number
	uniqueUsers: Set<string>
	totalDuration: number
	avgDuration: number
}

app.stream<string, PageView>(pageViewsTopic)
	.selectKey((_, v) => v.page)
	.groupByKey()
	.windowedBy(TimeWindows.of('15m'))
	.aggregate<{ views: number; users: string[]; totalDuration: number }>(
		() => ({ views: 0, users: [], totalDuration: 0 }),
		(page, view, agg) => ({
			views: agg.views + 1,
			users: agg.users.includes(view.userId) ? agg.users : [...agg.users, view.userId],
			totalDuration: agg.totalDuration + view.duration,
		})
	)
	.toStream()
	.mapValues((agg, windowedKey) => ({
		page: windowedKey.key,
		views: agg.views,
		uniqueUsers: agg.users.length,
		totalDuration: agg.totalDuration,
		avgDuration: agg.totalDuration / agg.views,
	}))
	.to(pageStatsTopic)
```

## Example: Running Totals

```typescript
interface Transaction {
	accountId: string
	amount: number
	type: 'credit' | 'debit'
}

interface AccountBalance {
	accountId: string
	balance: number
	transactionCount: number
}

app.stream<string, Transaction>(transactionsTopic)
	.groupByKey()
	.aggregate<AccountBalance>(
		() => ({ accountId: '', balance: 0, transactionCount: 0 }),
		(accountId, txn, agg) => ({
			accountId,
			balance: agg.balance + (txn.type === 'credit' ? txn.amount : -txn.amount),
			transactionCount: agg.transactionCount + 1,
		})
	)
	.toStream()
	.to(balancesTopic)
```

## Performance Considerations

1. **State size** - Aggregations maintain state per key
2. **Windowing** - Limits state by time
3. **Compaction** - Enable log compaction on output topics
4. **Serialization** - Use efficient codecs for aggregate values

```typescript
// Configure state store for better performance
const counts = stream.groupByKey().count({
	materialized: {
		storeName: 'counts',
		// Use LMDB for persistence
		storeProvider: lmdb({ stateDir: './state' }),
	},
})
```

---

---
url: https://chrisrecalis.github.io/kafkats/flow/joins.md
---
# Joins

Joins combine data from multiple streams or tables based on matching keys.

## Join Types

| Join  | Left Record | Right Record | Output              |
| ----- | ----------- | ------------ | ------------------- |
| Inner | Required    | Required     | When both present   |
| Left  | Required    | Optional     | When left present   |
| Outer | Optional    | Optional     | When either present |

## Join Options

Joins accept an optional `options` object (type `Joined<K, V1, V2>`):

| Option       | Type                                              | Description                                    |
| ------------ | ------------------------------------------------- | ---------------------------------------------- |
| `within`     | `TimeWindows \| SessionWindows \| SlidingWindows` | Join window (required for stream-stream joins) |
| `key`        | `Codec<K>`                                        | Key codec override                             |
| `value`      | `Codec<V1>`                                       | Left-side value codec override                 |
| `otherValue` | `Codec<V2>`                                       | Right-side value codec override                |

## Stream-Stream Joins

Join two streams within a time window:

```typescript
import { TimeWindows } from '@kafkats/flow'

const clicks = app.stream(clicksTopic)
const impressions = app.stream(impressionsTopic)

const clicksWithImpressions = clicks.join(
	impressions,
	(click, impression) => ({
		clickId: click.id,
		impressionId: impression.id,
		clicked: true,
	}),
	{ within: TimeWindows.of('5m') } // Events must be within 5 minutes
)
```

### Window Requirement

Stream-stream joins require a time window because streams are unbounded:

```typescript
// Join clicks and purchases within 1 hour
clicks.join(purchases, joiner, { within: TimeWindows.of('1h') })
```

### Join Types

```typescript
// Inner join - both must have matching event
clicks.join(purchases, joiner, { within: TimeWindows.of('1h') })

// Left join - emit for all clicks, purchase may be null
clicks.leftJoin(
	purchases,
	(click, purchase) => ({
		click,
		purchased: purchase !== null,
	}),
	{ within: TimeWindows.of('1h') }
)

// Outer join - emit for either click or purchase
clicks.outerJoin(
	purchases,
	(click, purchase) => ({
		click,
		purchase,
	}),
	{ within: TimeWindows.of('1h') }
)
```

## Stream-Table Joins

Join a stream with a table for enrichment:

```typescript
const orders = app.stream(ordersTopic)
const users = app.table(usersTopic)

const enrichedOrders = orders.join(users, (order, user) => ({
	orderId: order.id,
	userName: user.name,
	userEmail: user.email,
	total: order.total,
}))
```

### Key Matching

Both sides must have the same key for joining:

```typescript
// Orders keyed by orderId, users keyed by userId
// Need to rekey orders first
orders
	.selectKey((_, order) => order.userId) // Rekey by userId
	.join(users, joiner)
```

### Left Join

Keep all stream records even without table match:

```typescript
orders.leftJoin(users, (order, user) => ({
	orderId: order.id,
	userName: user?.name ?? 'Guest',
	total: order.total,
}))
```

## Table-Table Joins

Join two tables:

```typescript
const users = app.table(usersTopic)
const profiles = app.table(profilesTopic)

const fullUsers = users.join(profiles, (user, profile) => ({
	...user,
	...profile,
}))
```

### Changelog Semantics

Table-table joins update when either side changes:

```typescript
// When user updates → output updates
// When profile updates → output updates
```

## Global Table Joins

Join with a global table (fully replicated):

```typescript
const config = app.globalTable(configTopic)

app.stream(eventsTopic)
	.join(config, (event, configValue) => ({
		...event,
		setting: configValue.setting,
	}))
	.to(enrichedEventsTopic)
```

Global tables are useful for:

* Small, reference data
* Lookup tables needed everywhere
* Configuration data

## Co-partitioning Requirement

For joins to work correctly, both sides must be **co-partitioned**:

* Same number of partitions
* Same partitioning logic

```typescript
// Both topics must have same partition count and key type
const orders = topic('orders', { key: string(), ... })
const users = topic('users', { key: string(), ... })  // Same key type
```

If not co-partitioned, rekey through an intermediate topic:

```typescript
orders
	.selectKey((_, o) => o.userId)
	.through(rekeyedOrdersTopic) // Repartition
	.join(users, joiner)
```

## Example: Order Enrichment Pipeline

```typescript
interface Order {
	orderId: string
	userId: string
	productId: string
	quantity: number
	price: number
}

interface User {
	userId: string
	name: string
	tier: 'bronze' | 'silver' | 'gold'
}

interface Product {
	productId: string
	name: string
	category: string
}

interface EnrichedOrder {
	orderId: string
	userName: string
	userTier: string
	productName: string
	productCategory: string
	quantity: number
	price: number
	discount: number
}

// Tables for lookup
const users = app.table<string, User>(usersTopic)
const products = app.globalTable<string, Product>(productsTopic)

// Process orders
app.stream<string, Order>(ordersTopic)
	// First join with users (by userId)
	.selectKey((_, order) => order.userId)
	.join(users, (order, user) => ({ order, user }))

	// Then join with products (by productId)
	.selectKey((_, { order }) => order.productId)
	.join(
		products,
		({ order, user }, product) =>
			({
				orderId: order.orderId,
				userName: user.name,
				userTier: user.tier,
				productName: product.name,
				productCategory: product.category,
				quantity: order.quantity,
				price: order.price,
				discount: user.tier === 'gold' ? 0.15 : user.tier === 'silver' ? 0.1 : 0.05,
			}) as EnrichedOrder
	)

	.to(enrichedOrdersTopic)
```

## Example: Session Attribution

```typescript
import { TimeWindows } from '@kafkats/flow'

interface PageView {
	sessionId: string
	page: string
	timestamp: number
}

interface Conversion {
	sessionId: string
	product: string
	amount: number
	timestamp: number
}

interface Attribution {
	sessionId: string
	pages: string[]
	product: string
	amount: number
}

const pageViews = app.stream<string, PageView>(pageViewsTopic)
const conversions = app.stream<string, Conversion>(conversionsTopic)

// Attribute conversions to page views within 30 minutes
pageViews
	.groupByKey()
	.aggregate(
		() => ({ pages: [] as string[] }),
		(sessionId, pv, agg) => ({
			pages: [...agg.pages, pv.page],
		})
	)
	.toStream()
	.join(
		conversions,
		(pageViewAgg, conversion) =>
			({
				sessionId: conversion.sessionId,
				pages: pageViewAgg.pages,
				product: conversion.product,
				amount: conversion.amount,
			}) as Attribution,
		{ within: TimeWindows.of('30m') }
	)
	.to(attributionsTopic)
```

## Performance Tips

1. **Order matters** - Join smaller table to larger stream
2. **Use global tables** - For small lookup data
3. **Rekey sparingly** - Repartitioning is expensive
4. **Window size** - Smaller windows = less state
5. **Materialization** - Name stores for queryability

---

---
url: https://chrisrecalis.github.io/kafkats/flow/state-stores.md
---
# State Stores

State stores maintain the state needed for stateful operations like aggregations, joins, and windowing.

## State Store Types

| Store Type    | Use Case                                        |
| ------------- | ----------------------------------------------- |
| KeyValueStore | Simple key-value lookups (tables, aggregations) |
| WindowStore   | Time-windowed state (windowed aggregations)     |
| SessionStore  | Session-windowed state (session aggregations)   |

## Built-in: In-Memory Store

Default store for development and testing:

```typescript
import { flow, topic, inMemory } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
	stateStoreProvider: inMemory(), // Default
})
```

::: warning
In-memory stores are lost on restart. Use LMDB for persistence.
:::

## Persistent: LMDB Store

For production use, install the LMDB provider:

```bash
pnpm add @kafkats/flow-state-lmdb
```

```typescript
import { flow, topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'
import { lmdb } from '@kafkats/flow-state-lmdb'

const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
	stateStoreProvider: lmdb({
		stateDir: './state',
		mapSize: 1024 * 1024 * 1024, // 1GB max size
	}),
})
```

## Store Interfaces

### KeyValueStore

```typescript
interface KeyValueStore<K, V> {
	get(key: K): Promise<V | undefined>
	put(key: K, value: V): Promise<void>
	delete(key: K): Promise<void>
	all(): AsyncIterable<{ key: K; value: V }>
	range(from: K, to: K): AsyncIterable<{ key: K; value: V }>
}
```

### WindowStore

```typescript
interface WindowStore<K, V> {
	get(key: K, windowStart: number): Promise<V | undefined>
	put(key: K, value: V, windowStart: number): Promise<void>
	fetch(key: K, from: number, to: number): AsyncIterable<{ windowStart: number; value: V }>
	fetchAll(from: number, to: number): AsyncIterable<{ key: K; windowStart: number; value: V }>
}
```

### SessionStore

```typescript
interface SessionStore<K, V> {
	get(key: K, sessionStart: number, sessionEnd: number): Promise<V | undefined>
	put(key: K, value: V, sessionStart: number, sessionEnd: number): Promise<void>
	findSessions(
		key: K,
		from: number,
		to: number
	): AsyncIterable<{
		sessionStart: number
		sessionEnd: number
		value: V
	}>
}
```

## Materialization

Name stores for later querying:

```typescript
// Named store for counts
const counts = stream.groupByKey().count({
	materialized: { storeName: 'user-counts' },
})

// Query the store
const store = app.getStore<string, number>('user-counts')
const count = await store.get('user-123')
```

## Custom Store Provider

Implement your own store provider:

```typescript
interface StateStoreProvider {
	createKeyValueStore<K, V>(name: string, options: KeyValueStoreOptions<K, V>): KeyValueStore<K, V>

	createWindowStore<K, V>(name: string, options: WindowStoreOptions<K, V>): WindowStore<K, V>

	createSessionStore<K, V>(name: string, options: SessionStoreOptions<K, V>): SessionStore<K, V>

	close(): Promise<void>
}
```

Example Redis provider skeleton:

```typescript
import { StateStoreProvider, KeyValueStore } from '@kafkats/flow'
import Redis from 'ioredis'

class RedisStateStoreProvider implements StateStoreProvider {
	private redis: Redis

	constructor(options: { url: string }) {
		this.redis = new Redis(options.url)
	}

	createKeyValueStore<K, V>(name: string, options: KeyValueStoreOptions<K, V>) {
		return new RedisKeyValueStore<K, V>(this.redis, name, options)
	}

	// ... implement other methods

	async close() {
		await this.redis.quit()
	}
}
```

## Store Options

### KeyValueStoreOptions

| Option       | Type       | Required | Description                                |
| ------------ | ---------- | -------- | ------------------------------------------ |
| `keyCodec`   | `Codec<K>` | Yes      | Codec for serializing/deserializing keys   |
| `valueCodec` | `Codec<V>` | Yes      | Codec for serializing/deserializing values |

### WindowStoreOptions

| Option         | Type       | Required | Description                                |
| -------------- | ---------- | -------- | ------------------------------------------ |
| `keyCodec`     | `Codec<K>` | Yes      | Codec for keys                             |
| `valueCodec`   | `Codec<V>` | Yes      | Codec for values                           |
| `windowSizeMs` | `number`   | Yes      | Window size in milliseconds                |
| `retentionMs`  | `number`   | Yes      | How long to retain windows in milliseconds |

### SessionStoreOptions

| Option        | Type       | Required | Description                                 |
| ------------- | ---------- | -------- | ------------------------------------------- |
| `keyCodec`    | `Codec<K>` | Yes      | Codec for keys                              |
| `valueCodec`  | `Codec<V>` | Yes      | Codec for values                            |
| `retentionMs` | `number`   | Yes      | How long to retain sessions in milliseconds |

## Interactive Queries

Query state stores while the application runs:

```typescript
// Materialize the aggregation
const userCounts = stream.groupByKey().count({
	materialized: { storeName: 'user-counts' },
})

// Start the app
await app.start()

// Query the store
const store = app.getStore<string, number>('user-counts')

// HTTP endpoint example
app.get('/users/:id/count', async (req, res) => {
	const count = await store.get(req.params.id)
	res.json({ userId: req.params.id, count: count ?? 0 })
})
```

## Changelog Topics

State stores are backed by **changelog topics** for fault tolerance. Every state mutation is written to a Kafka topic, enabling state restoration after restarts or failures.

### Topic Naming

Changelog topics follow this naming convention:

```
{applicationId}-{storeName}-changelog
```

For example, an app with `applicationId: 'my-app'` and a store named `user-counts` creates:

```
my-app-user-counts-changelog
```

### Partition Count Inference

Changelog topics must have the **same number of partitions** as the source topic(s) to maintain data locality. When Task N processes partition N of the source topic, it must write to partition N of the changelog topic.

The partition count is automatically inferred:

```typescript
// Source topic has 8 partitions
// → Changelog topic created with 8 partitions
app.stream('orders', { key: codec.string(), value: codec.json<Order>() }).groupByKey().count() // Changelog: my-app-count-store-0-changelog (8 partitions)
```

For merged streams, the **maximum** partition count is used:

```typescript
const stream1 = app.stream('topic-a') // 4 partitions
const stream2 = app.stream('topic-b') // 8 partitions

stream1.merge(stream2).groupByKey().count() // Changelog created with 8 partitions (max)
```

### Validation

On startup, existing changelog topics are validated:

* If the changelog exists with the **correct** partition count → processing continues
* If the changelog exists with the **wrong** partition count → throws `ChangelogPartitionMismatchError`
* If the changelog doesn't exist → created automatically (unless `autoCreate: false`)

```typescript
import { ChangelogPartitionMismatchError, SourceTopicNotFoundError } from '@kafkats/flow'

try {
	await app.start()
} catch (err) {
	if (err instanceof ChangelogPartitionMismatchError) {
		console.error(`Changelog ${err.changelogTopic} has ${err.actualPartitions} partitions`)
		console.error(`Expected ${err.expectedPartitions} based on source topics: ${err.sourceTopics}`)
		// Fix: Delete the changelog topic and restart, or recreate with correct partition count
	}
	if (err instanceof SourceTopicNotFoundError) {
		console.error(`Source topic ${err.topic} doesn't exist for store ${err.storeName}`)
		// Fix: Create the source topic first
	}
}
```

### Global Configuration

Configure changelog behavior for all state stores:

```typescript
const app = flow({
	applicationId: 'my-app',
	client: { brokers: ['localhost:9092'] },
	changelog: {
		// Replication factor for all changelog topics
		replicationFactor: 3,

		// Additional topic configs applied to all changelogs
		topicConfigs: {
			'min.insync.replicas': '2',
			'segment.bytes': '104857600', // 100MB
		},

		// Set to false to skip auto-creation (production safety)
		autoCreate: false,
	},
})
```

::: tip Production Recommendation
Set `autoCreate: false` in production and pre-create changelog topics with your infrastructure tooling. This prevents accidental topic creation with incorrect settings.
:::

### Per-Store Configuration

Configure changelog settings for individual state stores:

```typescript
stream.groupByKey().count({
	storeName: 'user-counts',
	changelog: {
		// Custom topic name (default: {appId}-{storeName}-changelog)
		topicName: 'my-custom-changelog',

		// Replication factor for this store
		replicationFactor: 2,

		// Custom topic configs
		topicConfigs: {
			'retention.ms': '604800000', // 7 days
		},

		// Skip restoration on startup (not recommended)
		skipRestoration: false,
	},
})
```

### Disabling Changelogs

For ephemeral state that doesn't need persistence:

```typescript
stream.groupByKey().count({
	changelog: false, // No changelog topic created
})
```

::: warning
Without a changelog, state is lost on restart. Only disable for truly ephemeral computations.
:::

### Default Topic Configs

Changelog topics are created with these defaults (optimized for state stores):

| Config           | Value     | Description                        |
| ---------------- | --------- | ---------------------------------- |
| `cleanup.policy` | `compact` | Keep only latest value per key     |
| `retention.ms`   | `-1`      | Infinite retention                 |
| `segment.bytes`  | `50MB`    | Smaller segments for faster replay |

### State Restoration

On restart:

1. State store is restored from changelog topic
2. Consumer reads from beginning to end of changelog
3. Each record updates the local state store
4. Processing resumes from last committed offset

```
┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
│ Changelog Topic │────▶│ State Restoration│────▶│ Local Store     │
│ (Kafka)         │     │ Consumer         │     │ (Memory/LMDB)   │
└─────────────────┘     └──────────────────┘     └─────────────────┘
```

## Store Cleanup

Clean up state when processing:

```typescript
// Tombstone (null value) deletes from store
await producer.send(topic, [{ key: 'user-123', value: null }])
```

For windowed stores, old windows are automatically cleaned based on retention.

## Best Practices

1. **Name stores** - Use meaningful names for queryability
2. **Use LMDB in production** - For persistence and performance
3. **Configure retention** - Don't keep state forever
4. **Monitor size** - Watch state store disk usage
5. **Backup state** - Regularly back up LMDB directories
6. **Pre-create changelogs in production** - Use `autoCreate: false` and create topics via infrastructure tooling
7. **Match partition counts** - Ensure changelog partitions match source topic partitions
8. **Set replication factor** - Use `replicationFactor: 3` for production durability
9. **Handle startup errors** - Catch `ChangelogPartitionMismatchError` and `SourceTopicNotFoundError`

---

---
url: https://chrisrecalis.github.io/kafkats/flow/codecs.md
---
# Codecs

Codecs handle serialization and deserialization. They're provided by `@kafkats/client` and used throughout both client and flow packages.

## Built-in Codecs

### string

UTF-8 string encoding:

```typescript
import { string } from '@kafkats/client'

const codec = string()
```

### json

JSON serialization with TypeScript generics:

```typescript
import { json } from '@kafkats/client'

interface User {
	id: string
	name: string
}

const codec = json<User>()
```

### buffer

Raw buffer passthrough:

```typescript
import { buffer } from '@kafkats/client'

const codec = buffer()
```

## Using with Topics

```typescript
import { topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

const userEvents = topic('user-events', {
	key: string(),
	value: json<UserEvent>(),
})
```

## Custom Codecs

Create custom codecs with the `codec` function:

```typescript
import { codec } from '@kafkats/client'

const intCodec = codec<number>(
	n => {
		const buf = Buffer.alloc(4)
		buf.writeInt32BE(n)
		return buf
	},
	buf => buf.readInt32BE()
)
```

### Codec Interface

```typescript
interface Codec<T> {
	encode(value: T): Buffer
	decode(buffer: Buffer): T
}
```

## Use Cases

### In Topic Definitions

```typescript
import { topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

const orders = topic('orders', {
	key: string(),
	value: json<Order>(),
})

app.stream(orders)
	.filter((key, order) => order.total > 100)
	.to(outputTopic)
```

### In TestDriver

```typescript
import { TestDriver } from '@kafkats/flow/testing'
import { string, json } from '@kafkats/client'

const driver = new TestDriver()

driver.input('orders', {
	key: string(),
	value: json<Order>(),
})
```

### In State Stores

```typescript
import { string, json } from '@kafkats/client'

const store = provider.createKeyValueStore('my-store', {
	keyCodec: string(),
	valueCodec: json<MyValue>(),
})
```

## Protocol Buffers

```typescript
import { codec } from '@kafkats/client'
import { User } from './generated/user_pb.js'

const userCodec = codec<User>(
	user => Buffer.from(user.serializeBinary()),
	buf => User.deserializeBinary(buf)
)

const usersTopic = topic('users', {
	key: string(),
	value: userCodec,
})
```

## Avro

```typescript
import { codec } from '@kafkats/client'
import avro from 'avsc'

const userType = avro.Type.forSchema({
	type: 'record',
	name: 'User',
	fields: [
		{ name: 'id', type: 'string' },
		{ name: 'name', type: 'string' },
	],
})

const avroCodec = codec<{ id: string; name: string }>(
	user => userType.toBuffer(user),
	buf => userType.fromBuffer(buf)
)
```

## Zod Validation

Use [@kafkats/flow-codec-zod](/flow-codec-zod/) for runtime validation:

```typescript
import { zodCodec } from '@kafkats/flow-codec-zod'
import { z } from 'zod'

const UserSchema = z.object({
	id: z.string(),
	email: z.string().email(),
	age: z.number().min(0),
})

const userCodec = zodCodec(UserSchema)
// Validates on encode and decode
```

## Nullable Values

Handle null values in your codec:

```typescript
import { codec } from '@kafkats/client'

const nullableStringCodec = codec<string | null>(
	value => (value === null ? Buffer.alloc(0) : Buffer.from(value)),
	buf => (buf.length === 0 ? null : buf.toString())
)
```

## Composite Codecs

Combine multiple codecs:

```typescript
import { codec } from '@kafkats/client'

interface KeyValue {
	timestamp: number
	value: string
}

const kvCodec = codec<KeyValue>(
	kv => {
		const valueBuf = Buffer.from(kv.value)
		const buf = Buffer.alloc(8 + valueBuf.length)
		buf.writeBigInt64BE(BigInt(kv.timestamp))
		valueBuf.copy(buf, 8)
		return buf
	},
	buf => ({
		timestamp: Number(buf.readBigInt64BE()),
		value: buf.subarray(8).toString(),
	})
)
```

---

---
url: https://chrisrecalis.github.io/kafkats/flow/testing.md
---
# Testing

@kafkats/flow provides a testing module for writing tests without a real Kafka broker.

## Installation

The testing utilities are included in @kafkats/flow:

```typescript
import { TestDriver, ResultCollector } from '@kafkats/flow/testing'
```

## TestDriver

The main testing utility that mocks Kafka:

```typescript
import { TestDriver, ResultCollector } from '@kafkats/flow/testing'
import { string, json } from '@kafkats/client'

const driver = new TestDriver()

// Build topology
driver
	.input('orders', { key: string(), value: json<Order>() })
	.filter((_, order) => order.total > 100)
	.mapValues(order => ({ ...order, processed: true }))
	.to('large-orders', { value: json() })

// Run test
await driver.run(async ({ send, output }) => {
	await send('orders', { id: '1', total: 150 })
	await send('orders', { id: '2', total: 50 })

	const results = output('large-orders', { value: json() })
	expect(results).toHaveLength(1)
	expect(results[0].value.id).toBe('1')
})
```

## ResultCollector

Capture stream output without writing to a topic:

```typescript
const results = new ResultCollector<string, Order>()

driver
	.input('orders', { key: string(), value: json<Order>() })
	.filter((_, order) => order.total > 100)
	.peek(results.collector())

await driver.run(async ({ send }) => {
	await send('orders', { id: '1', total: 150 }, { key: 'order-1' })

	expect(results.values).toHaveLength(1)
	expect(results.first?.value.total).toBe(150)
	expect(results.first?.key).toBe('order-1')
})
```

### ResultCollector API

```typescript
interface ResultCollector<K, V> {
	// Get all collected records
	readonly records: Array<{ key: K; value: V }>

	// Get just values
	readonly values: V[]

	// Get just keys
	readonly keys: K[]

	// First/last record
	readonly first: { key: K; value: V } | undefined
	readonly last: { key: K; value: V } | undefined

	// Clear collected records
	clear(): void

	// Get collector function for peek()
	collector(): (key: K, value: V) => void
}
```

## Testing Tables

```typescript
const driver = new TestDriver()
const results = new ResultCollector<string, EnrichedEvent>()

// Create a table
const users = driver.table('users', {
	key: string(),
	value: json<User>(),
})

// Join stream with table
driver
	.input('events', { key: string(), value: json<Event>() })
	.join(users, (event, user) => ({
		...event,
		userName: user.name,
	}))
	.peek(results.collector())

await driver.run(async ({ send }) => {
	// Populate table first
	await send('users', { id: 'u1', name: 'Alice' }, { key: 'u1' })

	// Then send event
	await send('events', { action: 'click' }, { key: 'u1' })

	expect(results.first?.value.userName).toBe('Alice')
})
```

## Testing Windowed Aggregations

```typescript
import { TimeWindows } from '@kafkats/flow'

const driver = new TestDriver()
const results = new ResultCollector<string, number>()

driver
	.input('clicks', { key: string(), value: json() })
	.groupByKey()
	.windowedBy(TimeWindows.of('1h'))
	.count()
	.toStream()
	.peek((windowedKey, count) => {
		results.collector()(windowedKey.key, count)
	})

await driver.run(async ({ send }) => {
	const baseTime = Date.now()

	await send('clicks', {}, { key: 'user1', timestamp: baseTime })
	await send('clicks', {}, { key: 'user1', timestamp: baseTime + 1000 })
	await send('clicks', {}, { key: 'user1', timestamp: baseTime + 2000 })

	expect(results.last?.value).toBe(3)
})
```

## Test Utilities

### testRecord

Create a test record:

```typescript
import { testRecord } from '@kafkats/flow/testing'

const record = testRecord(
	'my-topic',
	'key',
	{ data: 'value' },
	{
		timestamp: Date.now(),
		headers: { 'trace-id': 'abc123' },
	}
)
```

### testRecordSequence

Create multiple records with incrementing timestamps:

```typescript
import { testRecordSequence } from '@kafkats/flow/testing'

const records = testRecordSequence(
	'my-topic',
	[
		{ key: 'k1', value: { n: 1 } },
		{ key: 'k2', value: { n: 2 } },
		{ key: 'k3', value: { n: 3 } },
	],
	{ baseTime: Date.now(), interval: 1000 }
)
```

### timestamps

Helper for time-based testing:

```typescript
import { timestamps } from '@kafkats/flow/testing'

const ts = timestamps(Date.now())

ts.now() // Current time
ts.plus('5m') // 5 minutes later
ts.minus('1h') // 1 hour earlier
ts.advance('10s') // Move forward 10 seconds
```

### quickTest

Minimal setup for simple tests:

```typescript
import { quickTest } from '@kafkats/flow/testing'

await quickTest(async ({ input, output }) => {
	input('numbers')
		.mapValues((n: number) => n * 2)
		.to('doubled')

	await input.send('numbers', 5)
	expect(output('doubled')[0]).toBe(10)
})
```

## Complete Example

```typescript
import { describe, it, expect } from 'vitest'
import { TestDriver, ResultCollector } from '@kafkats/flow/testing'
import { TimeWindows } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

interface ClickEvent {
	userId: string
	page: string
	timestamp: number
}

interface ClickStats {
	userId: string
	pageViews: number
	windowStart: number
}

describe('Click Analytics', () => {
	it('counts page views per user in 1-hour windows', async () => {
		const driver = new TestDriver()
		const results = new ResultCollector<string, ClickStats>()

		// Build topology
		driver
			.input('clicks', {
				key: string(),
				value: json<ClickEvent>(),
			})
			.groupByKey()
			.windowedBy(TimeWindows.of('1h'))
			.count()
			.toStream()
			.mapValues((count, windowedKey) => ({
				userId: windowedKey.key,
				pageViews: count,
				windowStart: windowedKey.window.start,
			}))
			.peek(results.collector())

		await driver.run(async ({ send }) => {
			const baseTime = new Date('2024-01-01T10:00:00Z').getTime()

			// User clicks
			await send('clicks', { userId: 'u1', page: '/home', timestamp: baseTime }, { key: 'u1' })
			await send('clicks', { userId: 'u1', page: '/products', timestamp: baseTime + 60000 }, { key: 'u1' })
			await send('clicks', { userId: 'u1', page: '/checkout', timestamp: baseTime + 120000 }, { key: 'u1' })

			// Check results
			expect(results.last?.value.pageViews).toBe(3)
			expect(results.last?.value.userId).toBe('u1')
		})
	})

	it('handles multiple users', async () => {
		const driver = new TestDriver()
		const results = new ResultCollector<string, number>()

		driver
			.input('clicks', { key: string(), value: json() })
			.groupByKey()
			.count()
			.toStream()
			.peek(results.collector())

		await driver.run(async ({ send }) => {
			await send('clicks', {}, { key: 'user1' })
			await send('clicks', {}, { key: 'user2' })
			await send('clicks', {}, { key: 'user1' })

			// Last update should show user1 with 2 clicks
			const user1Records = results.records.filter(r => r.key === 'user1')
			expect(user1Records[user1Records.length - 1]?.value).toBe(2)
		})
	})
})
```

---

---
url: https://chrisrecalis.github.io/kafkats/flow-codec-zod.md
---
# @kafkats/flow-codec-zod

Zod schema validation codecs for @kafkats/client and @kafkats/flow.

## Features

* **Runtime Validation** - Validate messages at encode and decode
* **Type Inference** - TypeScript types from Zod schemas
* **Error Messages** - Detailed validation error reporting
* **Zero Configuration** - Just pass your Zod schema

## Installation

```bash
pnpm add @kafkats/flow-codec-zod zod
```

## Quick Example

```typescript
import { flow, topic } from '@kafkats/flow'
import { string } from '@kafkats/client'
import { zodCodec } from '@kafkats/flow-codec-zod'
import { z } from 'zod'

// Define schema
const UserSchema = z.object({
	id: z.string().uuid(),
	email: z.string().email(),
	age: z.number().min(0).max(150),
	role: z.enum(['admin', 'user', 'guest']),
})

// Create codec from schema
const userCodec = zodCodec(UserSchema)

// Use in topic definition
const users = topic('users', {
	key: string(),
	value: userCodec,
})

// Type is inferred from schema
type User = z.infer<typeof UserSchema>
```

## How It Works

The codec:

1. **On encode**: Validates the value, then JSON stringifies
2. **On decode**: JSON parses, then validates the result

```typescript
// This will throw on invalid data
userCodec.encode({ id: 'not-a-uuid', email: 'invalid', age: -5, role: 'unknown' })
// ZodError: invalid_string at id, invalid_string at email, ...

// Valid data works
userCodec.encode({ id: '123e4567-e89b-12d3-a456-426614174000', email: 'a@b.com', age: 25, role: 'user' })
// Buffer containing JSON
```

## Next Steps

* [Usage Guide](/flow-codec-zod/usage) - Detailed usage patterns

---

---
url: https://chrisrecalis.github.io/kafkats/flow-codec-zod/usage.md
---
# Usage

## Basic Usage

```typescript
import { zodCodec } from '@kafkats/flow-codec-zod'
import { z } from 'zod'

// Simple schema
const MessageSchema = z.object({
	type: z.string(),
	payload: z.unknown(),
	timestamp: z.number(),
})

const codec = zodCodec(MessageSchema)
```

## With Flow Topics

```typescript
import { flow, topic } from '@kafkats/flow'
import { string } from '@kafkats/client'
import { zodCodec } from '@kafkats/flow-codec-zod'
import { z } from 'zod'

const OrderSchema = z.object({
	orderId: z.string(),
	userId: z.string(),
	items: z.array(
		z.object({
			productId: z.string(),
			quantity: z.number().int().positive(),
			price: z.number().positive(),
		})
	),
	total: z.number().positive(),
	status: z.enum(['pending', 'confirmed', 'shipped', 'delivered']),
})

const orders = topic('orders', {
	key: string(),
	value: zodCodec(OrderSchema),
})

const app = flow({
	applicationId: 'order-processor',
	client: { clientId: 'order-processor', brokers: ['localhost:9092'] },
})

// Types are fully inferred
app.stream(orders)
	.filter((_, order) => order.status === 'pending')
	.mapValues(order => ({ ...order, status: 'confirmed' as const }))
	.to(confirmedOrdersTopic)
```

## With Client Producer/Consumer

```typescript
import { KafkaClient, topic, string } from '@kafkats/client'
import { zodCodec } from '@kafkats/codec-zod'
import { z } from 'zod'

const EventSchema = z.object({
	type: z.string(),
	data: z.record(z.unknown()),
})

const events = topic('events', {
	key: string(),
	value: zodCodec(EventSchema),
})

const client = new KafkaClient({
	clientId: 'my-app',
	brokers: ['localhost:9092'],
})

// Producer - validates on send
const producer = client.producer()
await producer.send(events, [
	{
		key: 'event-1',
		value: { type: 'click', data: { page: '/home' } }, // Valid
	},
])

// Consumer - validates on receive
const consumer = client.consumer({ groupId: 'my-group' })
await consumer.runEach(events, async message => {
	// message.value is validated Event type
	console.log(message.value.type)
})
```

## Complex Schemas

### Nested Objects

```typescript
const AddressSchema = z.object({
	street: z.string(),
	city: z.string(),
	country: z.string(),
	zipCode: z.string(),
})

const CustomerSchema = z.object({
	id: z.string().uuid(),
	name: z.string().min(1),
	email: z.string().email(),
	addresses: z.array(AddressSchema),
	primaryAddressIndex: z.number().int().min(0),
})
```

### Unions and Discriminated Unions

```typescript
// Simple union
const IdSchema = z.union([z.string(), z.number()])

// Discriminated union (recommended)
const EventSchema = z.discriminatedUnion('type', [
	z.object({ type: z.literal('click'), page: z.string() }),
	z.object({ type: z.literal('purchase'), amount: z.number() }),
	z.object({ type: z.literal('signup'), email: z.string().email() }),
])

const eventCodec = zodCodec(EventSchema)
```

### Optional and Nullable

```typescript
const ProfileSchema = z.object({
	username: z.string(),
	bio: z.string().optional(),
	avatarUrl: z.string().url().nullable(),
	metadata: z.record(z.string()).default({}),
})
```

### Transformations

```typescript
const DateEventSchema = z.object({
	type: z.string(),
	// Transform string to Date on decode
	timestamp: z
		.string()
		.datetime()
		.transform(s => new Date(s)),
})

// Note: Transforms affect the inferred type
type DateEvent = z.infer<typeof DateEventSchema>
// { type: string; timestamp: Date }
```

## Error Handling

Zod throws detailed errors on validation failure:

```typescript
import { ZodError } from 'zod'

try {
	codec.decode(invalidBuffer)
} catch (error) {
	if (error instanceof ZodError) {
		console.log('Validation errors:')
		for (const issue of error.issues) {
			console.log(`  ${issue.path.join('.')}: ${issue.message}`)
		}
	}
}
```

### Graceful Error Handling

```typescript
// Create a codec that returns null on error
const safeCodec = {
	encode: (value: Order) => {
		const result = OrderSchema.safeParse(value)
		if (!result.success) {
			console.error('Encode failed:', result.error)
			return Buffer.alloc(0)
		}
		return Buffer.from(JSON.stringify(result.data))
	},
	decode: (buf: Buffer) => {
		try {
			const data = JSON.parse(buf.toString())
			const result = OrderSchema.safeParse(data)
			if (!result.success) {
				console.error('Decode failed:', result.error)
				return null
			}
			return result.data
		} catch {
			return null
		}
	},
}
```

## Schema Evolution

Handle schema changes gracefully:

```typescript
// V1 schema
const UserV1 = z.object({
	id: z.string(),
	name: z.string(),
})

// V2 schema with backward compatibility
const UserV2 = z.object({
	id: z.string(),
	name: z.string(),
	email: z.string().email().optional(), // New optional field
})

// V2 can decode V1 messages
```

## Performance Tips

1. **Reuse schemas** - Define schemas once, reuse everywhere
2. **Use `.strict()`** - Fail on extra properties
3. **Avoid heavy transforms** - Keep decode lightweight
4. **Consider caching** - For repeated validations

```typescript
// Good: Define once
const UserSchema = z.object({...}).strict()
const userCodec = zodCodec(UserSchema)

// Use everywhere
const topic1 = topic('users-v1', { value: userCodec })
const topic2 = topic('users-v2', { value: userCodec })
```

---

---
url: https://chrisrecalis.github.io/kafkats/flow-state-lmdb.md
---
# @kafkats/flow-state-lmdb

LMDB-backed persistent state stores for @kafkats/flow.

## Features

* **Persistent Storage** - State survives restarts
* **High Performance** - Memory-mapped, zero-copy reads
* **ACID Transactions** - Consistent state updates
* **Low Memory** - Data lives on disk, cached by OS
* **Battle-tested** - LMDB powers OpenLDAP, used by many projects

## Installation

```bash
pnpm add @kafkats/flow-state-lmdb
```

::: warning Native Dependencies
This package includes native bindings. Ensure you have build tools installed.
:::

## Quick Example

```typescript
import { flow } from '@kafkats/flow'
import { lmdb } from '@kafkats/flow-state-lmdb'

const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
	stateStoreProvider: lmdb({
		stateDir: './state',
	}),
})

// Aggregations now persist to disk
app.stream(clicks)
	.groupByKey()
	.count() // Stored in LMDB
	.toStream()
	.to(countsTopic)

await app.start()
```

## When to Use

| Use Case             | Recommendation      |
| -------------------- | ------------------- |
| Development          | In-memory (default) |
| Testing              | In-memory           |
| Production           | LMDB                |
| Large state          | LMDB                |
| Fast restarts needed | LMDB                |

## Store Types

LMDB provides all three store types:

* **LMDBKeyValueStore** - For tables and aggregations
* **LMDBWindowStore** - For windowed aggregations
* **LMDBSessionStore** - For session windows

## Next Steps

* [Configuration](/flow-state-lmdb/configuration) - Setup options
* [Store Types](/flow-state-lmdb/stores) - Detailed store documentation

---

---
url: https://chrisrecalis.github.io/kafkats/flow-state-lmdb/configuration.md
---
# Configuration

## Basic Setup

```typescript
import { flow } from '@kafkats/flow'
import { lmdb } from '@kafkats/flow-state-lmdb'

const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
	stateStoreProvider: lmdb({
		stateDir: './state',
	}),
})
```

## Options

```typescript
lmdb({
	// Required: Directory for LMDB files
	stateDir: './state',

	// Optional: Maximum database size (default: 1GB)
	mapSize: 1024 * 1024 * 1024,

	// Optional: Maximum number of named databases (default: 100)
	maxDbs: 100,
})
```

| Option     | Type     | Default              | Description                       |
| ---------- | -------- | -------------------- | --------------------------------- |
| `stateDir` | `string` | -                    | Directory for LMDB database files |
| `mapSize`  | `number` | `1024 * 1024 * 1024` | Maximum database size in bytes    |
| `maxDbs`   | `number` | `100`                | Maximum number of named databases |

### stateDir

Directory where LMDB stores its files:

```typescript
lmdb({ stateDir: './data/kafka-state' })
```

The directory will be created if it doesn't exist. Structure:

```
./data/kafka-state/
├── data.mdb      # Main database file
└── lock.mdb      # Lock file
```

### mapSize

Maximum size of the database. LMDB pre-allocates virtual address space:

```typescript
// 1GB (default)
lmdb({ stateDir: './state', mapSize: 1024 * 1024 * 1024 })

// 10GB for larger state
lmdb({ stateDir: './state', mapSize: 10 * 1024 * 1024 * 1024 })

// 100GB for very large state
lmdb({ stateDir: './state', mapSize: 100n * 1024n * 1024n * 1024n })
```

::: tip
Start with a generous mapSize. It's virtual memory, not actual disk usage. Increasing it later requires a restart.
:::

### maxDbs

Maximum number of named databases (stores):

```typescript
// 100 databases (default)
lmdb({ stateDir: './state', maxDbs: 100 })

// More for complex topologies
lmdb({ stateDir: './state', maxDbs: 500 })
```

Each materialized store uses one database. Count your:

* KTable materializations
* Aggregation results
* Window stores
* Session stores

## Environment-Based Configuration

```typescript
import { flow, inMemory } from '@kafkats/flow'
import { lmdb } from '@kafkats/flow-state-lmdb'

const stateProvider =
	process.env.NODE_ENV === 'production'
		? lmdb({
				stateDir: process.env.STATE_DIR || '/var/lib/kafka-state',
				mapSize: 10 * 1024 * 1024 * 1024,
			})
		: inMemory()

const app = flow({
	applicationId: 'my-app',
	client: { clientId: 'my-app', brokers: ['localhost:9092'] },
	stateStoreProvider: stateProvider,
})
```

## Directory Structure

Organize state by application:

```
/var/lib/kafka-state/
├── order-processor/
│   ├── data.mdb
│   └── lock.mdb
├── user-analytics/
│   ├── data.mdb
│   └── lock.mdb
└── inventory-tracker/
    ├── data.mdb
    └── lock.mdb
```

```typescript
const app = flow({
	applicationId: 'order-processor',
	client: { clientId: 'order-processor', brokers: ['localhost:9092'] },
	stateStoreProvider: lmdb({
		stateDir: `/var/lib/kafka-state/${applicationId}`,
	}),
})
```

## Docker Configuration

Mount a volume for state persistence:

```yaml
# docker-compose.yml
services:
    stream-processor:
        image: my-app:latest
        volumes:
            - kafka-state:/var/lib/kafka-state
        environment:
            - STATE_DIR=/var/lib/kafka-state

volumes:
    kafka-state:
```

## Kubernetes Configuration

Use a PersistentVolumeClaim:

```yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
    name: kafka-state-pvc
spec:
    accessModes:
        - ReadWriteOnce
    resources:
        requests:
            storage: 10Gi
---
apiVersion: apps/v1
kind: Deployment
metadata:
    name: stream-processor
spec:
    replicas: 1 # Must be 1 for RWO PVC
    template:
        spec:
            containers:
                - name: app
                  volumeMounts:
                      - name: state
                        mountPath: /var/lib/kafka-state
            volumes:
                - name: state
                  persistentVolumeClaim:
                      claimName: kafka-state-pvc
```

::: warning Single Writer
LMDB supports single-writer, multiple-reader. Don't run multiple instances with the same stateDir.
:::

## Cleanup

Remove old state:

```bash
# Stop the application first
rm -rf ./state/data.mdb ./state/lock.mdb
```

State will be rebuilt from Kafka changelog topics on next start.

---

---
url: https://chrisrecalis.github.io/kafkats/flow-state-lmdb/stores.md
---
# Store Types

@kafkats/flow-state-lmdb provides three store types matching the flow state interfaces.

## LMDBKeyValueStore

For simple key-value storage (tables, aggregations):

```typescript
interface KeyValueStore<K, V> {
	get(key: K): Promise<V | undefined>
	put(key: K, value: V): Promise<void>
	delete(key: K): Promise<void>
	all(): AsyncIterable<{ key: K; value: V }>
	range(from: K, to: K): AsyncIterable<{ key: K; value: V }>
}
```

### Use Cases

* Table materializations
* Non-windowed aggregations
* Lookup data

### Key Format

Keys are stored as-is using the key codec:

```
[encoded_key] → [encoded_value]
```

## LMDBWindowStore

For time-windowed state:

```typescript
interface WindowStore<K, V> {
	get(key: K, windowStart: number): Promise<V | undefined>
	put(key: K, value: V, windowStart: number): Promise<void>
	fetch(
		key: K,
		from: number,
		to: number
	): AsyncIterable<{
		windowStart: number
		value: V
	}>
	fetchAll(
		from: number,
		to: number
	): AsyncIterable<{
		key: K
		windowStart: number
		value: V
	}>
}
```

### Use Cases

* Tumbling window aggregations
* Hopping window aggregations
* Sliding window aggregations

### Key Format

Keys are ordered for efficient window queries:

```
[windowStart:8bytes][windowEnd:8bytes][key] → [value]
```

This ordering enables:

* Efficient range scans by time
* Fast lookups for specific key + window
* Ordered iteration by window time

## LMDBSessionStore

For session-windowed state:

```typescript
interface SessionStore<K, V> {
	get(key: K, sessionStart: number, sessionEnd: number): Promise<V | undefined>
	put(key: K, value: V, sessionStart: number, sessionEnd: number): Promise<void>
	findSessions(
		key: K,
		from: number,
		to: number
	): AsyncIterable<{
		sessionStart: number
		sessionEnd: number
		value: V
	}>
}
```

### Use Cases

* Session window aggregations
* Activity-based grouping

### Key Format

Keys are ordered by key first, then time:

```
[key][sessionStart:8bytes][sessionEnd:8bytes] → [value]
```

This ordering enables:

* Fast session lookup for a key
* Efficient merging of adjacent sessions
* Range queries for a key's sessions

## Store Creation

Stores are created automatically by the flow runtime:

```typescript
// This creates an LMDBKeyValueStore internally
const counts = stream.groupByKey().count({
	materialized: { storeName: 'user-counts' },
})

// This creates an LMDBWindowStore internally
const windowedCounts = stream
	.groupByKey()
	.windowedBy(TimeWindows.of('1h'))
	.count({
		materialized: { storeName: 'hourly-counts' },
	})
```

## Manual Store Access

Create stores directly for custom use:

```typescript
import { lmdb } from '@kafkats/flow-state-lmdb'
import { string, json } from '@kafkats/flow'

const provider = lmdb({ stateDir: './state' })

// KeyValue store
const kvStore = provider.createKeyValueStore('my-kv', {
	keyCodec: string(),
	valueCodec: json<MyValue>(),
})

await kvStore.put('key1', { data: 'value1' })
const value = await kvStore.get('key1')

// Window store
const windowStore = provider.createWindowStore('my-windows', {
	keyCodec: string(),
	valueCodec: json<number>(),
	windowSize: 3600000, // 1 hour in ms
})

const windowStart = Math.floor(Date.now() / 3600000) * 3600000
await windowStore.put('user1', 42, windowStart)

// Cleanup
await provider.close()
```

## Performance Characteristics

### LMDBKeyValueStore

| Operation | Complexity   | Notes           |
| --------- | ------------ | --------------- |
| get       | O(log n)     | B+ tree lookup  |
| put       | O(log n)     | Single write    |
| delete    | O(log n)     | Tombstone write |
| all       | O(n)         | Full scan       |
| range     | O(log n + k) | k = results     |

### LMDBWindowStore

| Operation | Complexity   | Notes                |
| --------- | ------------ | -------------------- |
| get       | O(log n)     | Composite key lookup |
| put       | O(log n)     | Single write         |
| fetch     | O(log n + k) | Time-range scan      |
| fetchAll  | O(log n + k) | Time-range scan      |

### LMDBSessionStore

| Operation    | Complexity   | Notes                |
| ------------ | ------------ | -------------------- |
| get          | O(log n)     | Composite key lookup |
| put          | O(log n)     | May merge sessions   |
| findSessions | O(log n + k) | Key + time scan      |

## Memory Usage

LMDB is memory-mapped:

* **Reads**: OS pages data into memory on demand
* **Writes**: Buffered then synced to disk
* **Cache**: OS manages the page cache

Monitor with:

```bash
# Database size
du -sh ./state/data.mdb

# Memory-mapped usage
cat /proc/$(pgrep -f my-app)/maps | grep data.mdb
```

## Durability

LMDB provides:

* **Atomic commits**: All or nothing
* **Crash safety**: Never corrupted on crash
* **Sync writes**: `MDB_NOSYNC` is NOT used

Data is durable immediately after `put()` returns.

## Compaction

LMDB does not automatically compact. To reclaim space:

```bash
# 1. Stop the application
# 2. Copy database with mdb_copy
mdb_copy -c ./state/data.mdb ./state/compacted.mdb

# 3. Replace original
mv ./state/compacted.mdb ./state/data.mdb

# 4. Restart application
```

For production, schedule periodic compaction during low-traffic periods.

---

---
url: https://chrisrecalis.github.io/kafkats/examples.md
---
# Examples

Complete, working examples to help you get started with kafkats.

## Basic Examples

| Example                                      | Description                         |
| -------------------------------------------- | ----------------------------------- |
| [Simple Producer](/examples/simple-producer) | Send messages to Kafka              |
| [Consumer Group](/examples/consumer-group)   | Read messages with a consumer group |

## Stream Processing

| Example                                          | Description                         |
| ------------------------------------------------ | ----------------------------------- |
| [Stream Processing](/examples/stream-processing) | Basic stream transformations        |
| [Word Count](/examples/word-count)               | Classic word count with aggregation |

## Running the Examples

1. Start Kafka locally:

```bash
docker run -d --name kafka \
  -p 9092:9092 \
  -e KAFKA_CFG_NODE_ID=0 \
  -e KAFKA_CFG_PROCESS_ROLES=controller,broker \
  -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@localhost:9093 \
  -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  bitnami/kafka:latest
```

2. Install dependencies:

```bash
pnpm add @kafkats/client @kafkats/flow
```

3. Run the example:

```bash
npx tsx example.ts
```

---

---
url: https://chrisrecalis.github.io/kafkats/examples/simple-producer.md
---
# Simple Producer

A basic example of producing messages to Kafka.

## Code

```typescript
import { KafkaClient, topic, string, json } from '@kafkats/client'

// Define a typed topic
interface UserEvent {
	userId: string
	action: 'login' | 'logout' | 'signup'
	timestamp: number
}

const userEvents = topic('user-events', {
	key: string(),
	value: json<UserEvent>(),
})

async function main() {
	// Create client
	const client = new KafkaClient({
		clientId: 'simple-producer',
		brokers: ['localhost:9092'],
	})

	// Create producer
	const producer = client.producer({
		acks: 'all', // Wait for all replicas
		compression: 'snappy',
	})

	try {
		// Send some events
		const events: Array<{ key: string; value: UserEvent }> = [
			{
				key: 'user-1',
				value: { userId: 'user-1', action: 'login', timestamp: Date.now() },
			},
			{
				key: 'user-2',
				value: { userId: 'user-2', action: 'signup', timestamp: Date.now() },
			},
			{
				key: 'user-1',
				value: { userId: 'user-1', action: 'logout', timestamp: Date.now() + 1000 },
			},
		]

		// Send messages
		const results = await producer.send(userEvents, events)

		// Print results
		for (const result of results) {
			console.log(`Sent to ${result.topic}[${result.partition}] @ offset ${result.offset}`)
		}

		// Ensure all messages are sent
		await producer.flush()
		console.log('All messages sent!')
	} finally {
		// Close producer
		await producer.disconnect()
	}
}

main().catch(console.error)
```

## Output

```
Sent to user-events[0] @ offset 0
Sent to user-events[0] @ offset 1
Sent to user-events[0] @ offset 2
All messages sent!
```

## Key Points

1. **Typed topics** - Use `topic()` with codecs for type safety
2. **Acknowledgments** - `acks: 'all'` ensures durability
3. **Compression** - Reduces network bandwidth
4. **Flush** - Ensures all batched messages are sent
5. **Disconnect** - Always disconnect the producer when done

## Variations

### Fire and Forget

Don't wait for acknowledgments (fastest, least reliable):

```typescript
const producer = client.producer({
	acks: 'none',
})
```

### With Headers

Add metadata to messages:

```typescript
await producer.send(userEvents, [
	{
		key: 'user-1',
		value: { userId: 'user-1', action: 'login', timestamp: Date.now() },
		headers: {
			'trace-id': 'abc123',
			source: 'web-app',
		},
	},
])
```

### Multiple Topics

Send to different topics with separate calls:

```typescript
await producer.send('events', [{ key: 'k1', value: JSON.stringify({ type: 'event' }) }])
await producer.send('logs', [{ key: 'k2', value: JSON.stringify({ type: 'log' }) }])
```

---

---
url: https://chrisrecalis.github.io/kafkats/examples/consumer-group.md
---
# Consumer Group

Read messages from Kafka using a consumer group.

## Code

```typescript
import { KafkaClient, topic, string, json } from '@kafkats/client'

interface UserEvent {
	userId: string
	action: string
	timestamp: number
}

const userEvents = topic('user-events', {
	key: string(),
	value: json<UserEvent>(),
})

async function main() {
	const client = new KafkaClient({
		clientId: 'consumer-example',
		brokers: ['localhost:9092'],
	})

	const consumer = client.consumer({
		groupId: 'user-events-processor',
		autoOffsetReset: 'earliest', // Start from beginning
	})

	// Handle shutdown
	const controller = new AbortController()

	process.on('SIGINT', () => {
		console.log('\nShutting down...')
		controller.abort()
	})

	process.on('SIGTERM', () => {
		console.log('\nShutting down...')
		controller.abort()
	})

	try {
		console.log('Waiting for messages...')

		// Process messages
		await consumer.runEach(
			userEvents,
			async (message, ctx) => {
				console.log(`[${ctx.topic}:${ctx.partition}] offset=${ctx.offset}`)
				console.log(`  Key: ${message.key}`)
				console.log(`  Value: ${JSON.stringify(message.value)}`)
				console.log()
			},
			{
				signal: controller.signal,
			}
		)
	} finally {
		consumer.stop()
		console.log('Consumer closed')
	}
}

main().catch(console.error)
```

## Output

```
Waiting for messages...
[user-events:0] offset=0
  Key: user-1
  Value: {"userId":"user-1","action":"login","timestamp":1703001234567}

[user-events:0] offset=1
  Key: user-2
  Value: {"userId":"user-2","action":"signup","timestamp":1703001234567}

[user-events:0] offset=2
  Key: user-1
  Value: {"userId":"user-1","action":"logout","timestamp":1703001235567}

^C
Shutting down...
Consumer closed
```

## Key Points

1. **Consumer group** - Multiple consumers share the workload
2. **Auto offset reset** - Start from `earliest` or `latest`
3. **Typed messages** - Using topic definition with codecs
4. **Graceful shutdown** - Handle SIGINT/SIGTERM
5. **Context** - Access topic, partition, offset metadata

## Variations

### Batch Processing

Process multiple messages at once:

```typescript
await consumer.runBatch(
	userEvents,
	async (messages, ctx) => {
		console.log(`Received ${messages.length} messages`)

		for (const message of messages) {
			await processMessage(message)
		}
	},
	{
		maxBatchSize: 100,
		maxBatchWaitMs: 50,
	}
)
```

### Parallel Partition Processing

Process multiple partitions concurrently:

```typescript
await consumer.runEach(userEvents, handler, {
	partitionConcurrency: 4, // Process 4 partitions in parallel
})
```

### Multiple Topics

Subscribe to multiple topics:

```typescript
await consumer.runEach(['events', 'logs', 'metrics'], handler)
```

---

---
url: https://chrisrecalis.github.io/kafkats/examples/stream-processing.md
---
# Stream Processing

Basic stream transformations with @kafkats/flow.

## Code

```typescript
import { flow, topic } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

// Input event
interface RawEvent {
	type: string
	userId: string
	data: Record<string, unknown>
	timestamp: number
}

// Output event
interface ProcessedEvent {
	type: string
	userId: string
	data: Record<string, unknown>
	processedAt: number
	source: string
}

// Define topics
const rawEvents = topic('raw-events', {
	key: string(),
	value: json<RawEvent>(),
})

const processedEvents = topic('processed-events', {
	key: string(),
	value: json<ProcessedEvent>(),
})

const errorEvents = topic('error-events', {
	key: string(),
	value: json<RawEvent>(),
})

async function main() {
	const app = flow({
		applicationId: 'event-processor',
		client: { clientId: 'event-processor', brokers: ['localhost:9092'] },
	})

	// Build processing topology
	const [valid, invalid] = app
		.stream(rawEvents)
		// Log incoming events
		.peek((key, event) => {
			console.log(`Received: ${event.type} from ${event.userId}`)
		})
		// Split valid and invalid events
		.branch(
			(_, event) => event.type !== 'heartbeat' && event.userId.length > 0,
			() => true
		)

	// Process valid events
	valid
		.mapValues(
			event =>
				({
					type: event.type,
					userId: event.userId,
					data: event.data,
					processedAt: Date.now(),
					source: 'event-processor',
				}) as ProcessedEvent
		)
		.peek((key, event) => {
			console.log(`Processed: ${event.type}`)
		})
		.to(processedEvents)

	// Route invalid events to error topic
	invalid
		.peek((key, event) => {
			console.log(`Invalid event: ${JSON.stringify(event)}`)
		})
		.to(errorEvents)

	// Handle shutdown
	process.on('SIGTERM', async () => {
		console.log('Shutting down...')
		await app.close()
	})

	// Start processing
	console.log('Starting stream processor...')
	await app.start()
}

main().catch(console.error)
```

## How It Works

1. **Read** from `raw-events` topic
2. **Peek** to log each event (side effect)
3. **Branch** into valid and invalid streams
4. **Transform** valid events with processing metadata
5. **Write** to respective output topics

## Topology Visualization

```
raw-events
    │
    ├──► peek (log)
    │
    ├──► branch
    │       │
    │       ├── valid ──► mapValues ──► peek ──► processed-events
    │       │
    │       └── invalid ──► peek ──► error-events
```

## Testing the Example

1. Start the processor:

```bash
npx tsx stream-processing.ts
```

2. In another terminal, produce some events:

```typescript
import { KafkaClient, topic, string, json } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'test-producer',
	brokers: ['localhost:9092'],
})

const producer = client.producer()

await producer.send('raw-events', [
	{
		key: 'user-1',
		value: JSON.stringify({
			type: 'click',
			userId: 'user-1',
			data: { page: '/home' },
			timestamp: Date.now(),
		}),
	},
	{
		key: 'heartbeat',
		value: JSON.stringify({
			type: 'heartbeat',
			userId: 'user-1',
			data: {},
			timestamp: Date.now(),
		}),
	},
])

await producer.disconnect()
```

3. Check the output topics:

```bash
# Processed events (valid)
kafka-console-consumer --topic processed-events --from-beginning

# Error events (invalid)
kafka-console-consumer --topic error-events --from-beginning
```

---

---
url: https://chrisrecalis.github.io/kafkats/examples/word-count.md
---
# Word Count

The classic word count example using @kafkats/flow.

## Code

```typescript
import { flow, topic, TimeWindows } from '@kafkats/flow'
import { string, json } from '@kafkats/client'

// Input: lines of text
const lines = topic('lines', {
	key: string(),
	value: string(),
})

// Output: word counts
interface WordCount {
	word: string
	count: number
	windowStart: number
	windowEnd: number
}

const wordCounts = topic('word-counts', {
	key: string(),
	value: json<WordCount>(),
})

async function main() {
	const app = flow({
		applicationId: 'word-count',
		client: { clientId: 'word-count', brokers: ['localhost:9092'] },
	})

	app.stream(lines)
		// Split lines into words
		.flatMapValues(line =>
			line
				.toLowerCase()
				.split(/\s+/)
				.filter(word => word.length > 0)
		)

		// Rekey by word
		.selectKey((_, word) => word)

		// Group by word
		.groupByKey()

		// Count in 1-minute windows
		.windowedBy(TimeWindows.of('1m'))
		.count()

		// Convert to output format
		.toStream()
		.map((windowedKey, count) => ({
			key: windowedKey.key,
			value: {
				word: windowedKey.key,
				count,
				windowStart: windowedKey.window.start,
				windowEnd: windowedKey.window.end,
			},
		}))

		// Write results
		.to(wordCounts)

	// Handle shutdown
	process.on('SIGTERM', async () => {
		await app.close()
	})

	console.log('Word count processor started')
	await app.start()
}

main().catch(console.error)
```

## How It Works

1. **Read** lines of text from input topic
2. **FlatMapValues** - Split each line into words
3. **SelectKey** - Rekey by the word itself
4. **GroupByKey** - Group all occurrences of each word
5. **WindowedBy** - Apply 1-minute time windows
6. **Count** - Count occurrences per word per window
7. **Map** - Transform to output format
8. **To** - Write to output topic

## Topology Visualization

```
lines
  │
  ├──► flatMapValues (split into words)
  │
  ├──► selectKey (key by word)
  │
  ├──► groupByKey
  │
  ├──► windowedBy (1 minute)
  │
  ├──► count
  │
  ├──► toStream
  │
  ├──► map (format output)
  │
  └──► word-counts
```

## Testing

1. Start the word count processor:

```bash
npx tsx word-count.ts
```

2. Send some text:

```typescript
import { KafkaClient } from '@kafkats/client'

const client = new KafkaClient({
	clientId: 'producer',
	brokers: ['localhost:9092'],
})

const producer = client.producer()

await producer.send('lines', [
	{ key: 'doc1', value: 'hello world hello' },
	{ key: 'doc2', value: 'hello kafka streams' },
	{ key: 'doc3', value: 'kafka is great kafka' },
])

await producer.disconnect()
```

3. Check results:

```bash
kafka-console-consumer --topic word-counts --from-beginning
```

Expected output:

```json
{"word":"hello","count":1,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"world","count":1,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"hello","count":2,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"hello","count":3,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"kafka","count":1,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"streams","count":1,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"kafka","count":2,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"is","count":1,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"great","count":1,"windowStart":1703001200000,"windowEnd":1703001260000}
{"word":"kafka","count":3,"windowStart":1703001200000,"windowEnd":1703001260000}
```

## Variations

### Global Count (No Window)

Count all time:

```typescript
app.stream(lines)
	.flatMapValues(line => line.toLowerCase().split(/\s+/))
	.selectKey((_, word) => word)
	.groupByKey()
	.count() // No windowing
	.toStream()
	.to(wordCounts)
```

### Top N Words

Find top 10 words per window:

```typescript
// Would require additional processing:
// 1. Collect all word counts
// 2. Sort by count
// 3. Take top 10
```

### Stop Words Filtering

Filter common words:

```typescript
const stopWords = new Set(['the', 'a', 'an', 'is', 'are', 'was', 'were'])

app.stream(lines).flatMapValues(line =>
	line
		.toLowerCase()
		.split(/\s+/)
		.filter(word => word.length > 0 && !stopWords.has(word))
)
// ... rest of pipeline
```