Consumer API

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

Creating a Consumer

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:

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:

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

Each batch holds records from one partition, up to the consumer's maxBytesPerPartition (estimated from key and value sizes). Records fetched ahead while a handler is busy stay buffered for the next batch, so a slow handler never receives more than one partition's byte budget at once. A record larger than the limit is delivered on its own. Lower maxBytesPerPartition when per-record work is slow: a rebalance waits for in-flight batches, so smaller batches keep rebalances and topic priority responsive (smaller fetches are the trade-off).

Async Iterator Mode (stream)

Consume messages via for await ... of:

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

Message Structure

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

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:

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

Typed Consumers

Use typed topics for automatic deserialization:

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
priority	PriorityStrategy	-	Prioritize topics when fetching and delivering

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
assignment	ManualAssignment[]	-	Manually assign partitions instead of joining group
priority	PriorityStrategy	-	Prioritize topics when fetching and delivering

Partition Concurrency

Control how many partitions are processed concurrently:

// 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.

Topic Priority

When one consumer handles latency-sensitive and backlog topics, use a priority strategy to control both which buffered records are delivered and which partitions the background fetcher requests.

import { strict, weighted } from '@kafkats/client'

const live = topic('live')
const bulk = topic('bulk')

await consumer.runEach([live, bulk], handler, {
	partitionConcurrency: 4,
	priority: strict({ order: ['live', 'bulk'] }),
})

// Reserve an approximate 80/20 share while allowing either topic to use spare capacity.
await consumer.runEach([live, bulk], handler, {
	priority: weighted({ shares: { live: 8, bulk: 2 } }),
})

strict can starve lower-priority topics while higher-priority traffic remains available. Use weighted when every topic must continue making progress. Topics omitted from a strict order share the final tier; topics omitted from weighted shares receive a share of 1.

Custom strategies implement PriorityStrategy and return ordered drain entries, each capped at maxBytes out of the poll's byteBudget (the consumer's maxBytesPerPartition), plus the partitions currently allowed to fetch. Buffered partitions omitted from drain are still delivered FIFO after the planned entries within the remaining poll budget, so hold a topic back through fetch, not drain. Derive fairness from the supplied scheduler state: the fetch loop may call schedule() without a drain, and a drain decision may be only partially applied.

Priority applies at poll granularity. A handler batch already admitted is never preempted; lower maxBytesPerPartition for finer-grained scheduling when partitionConcurrency is small. Literal topic names are checked against the subscription. One TypeScript limitation is that topic<Order>('orders') leaves the name typed as string because generic arguments cannot be partially inferred; codec-driven inference preserves the literal name.

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

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:

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:

// 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:

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:

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:

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:

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

Basic Usage

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

With Explicit Starting Offset

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:

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

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

With abort signal:

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

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:

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

Next Steps

• Error Handling - Error types and recovery
• Codecs - Custom serialization
• Transactions - Exactly-once consume-transform-produce