Getting Started with @kafkats/flow

Installation

pnpm add @kafkats/flow

Creating a Flow Application

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.

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.

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:

const consumer = client.consumer({
	groupId: 'downstream-consumer',
	isolationLevel: 'read_committed',
})

Defining Topics

Define typed topics for input and output:

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

// Create a stream from a topic
app.stream(events)
	.filter((key, value) => value.action === 'purchase')
	.mapValues(value => ({ ...value, processed: true }))
	.to(outputTopic)

Table Processing

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

app.stream(events).groupByKey().count().toStream().to(countsTopic)

Starting and Stopping

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

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 - Stream transformations
• KTable Operations - Table transformations
• Aggregations - Counting and reducing
• Windowing - Time-based processing