KafkaRouter

KafkaRouter routes Apache Kafka messages to handlers, one record at a time. It takes both Amazon MSK and self-managed clusters.

A single event can carry many records, spread across more than one topic and partition. The router base64 decodes each record's key and value, parses the value as JSON, decodes the headers to text, then works out which of your routes should handle it. Records run in the order they arrive rather than in parallel, because a Kafka partition is ordered.

Install

bash

npm install @lambda-event-router/base @lambda-event-router/kafka

@lambda-event-router/base is a peer dependency, so install it yourself. It exports LambdaRouter, which every router plugs into.

Create the router

import { createKafkaRouter } from '@lambda-event-router/kafka'
import { logInvocation } from './middleware/logInvocation'

const kafkaRouter = createKafkaRouter({
  batchItemFailures: true,  // Optional
  middleware: [logInvocation],  // Optional
})

Both options can be left out. createKafkaRouter() on its own gives you a router that fails the whole batch when any record throws.

Options

Option	Type	Required	Default	Description
`batchItemFailures`	`boolean`	No	`false`	Report failed records back to Lambda instead of failing the whole batch. See Failures and retries
`middleware`	`KafkaMiddleware[]`	No	`[]`	Runs for every record this router handles, before any route middleware. See Middleware

Register routes

kafkaRouter.route({
  filters: {
    topic: STOCK_TOPIC,
    eventSourceArn: MSK_CLUSTER_ARN,
  },
  valueSchema: StockMovementSchema,  // Optional
  middleware: [withWarehouseContext],  // Optional
  handler: onStockMoved,
})

filters and handler are the only required keys.

route() returns the router, so you can chain registrations.

kafkaRouter.route(stockMovedRoute).route(stockAlertRoute)

Routes are ranked by how specific they are. Where two overlap, the one matching only a subset of the other is tried first, whatever order you registered them in. Registration order decides the rest, so give each route filters no other route can match. See match order for how ranking works and what it cannot settle.

A record that matches no route throws. With batchItemFailures off that fails the entire batch, including records that would have succeeded, so adding a topic to the event source mapping without adding a route for it takes the whole invocation down. Register a catch-all route with empty filters if you would rather swallow unknown records, and see nothing matched for what the other routers do instead.

Filters

Every filter key on one route, showing each form a value can take. All of them are optional, so set the ones that pick out the records you want and leave the rest off.

kafkaRouter.route({
  filters: {
    topic: [STOCK_TOPIC, ALERT_TOPIC],
    eventSourceArn: MSK_CLUSTER_ARN,
    bootstrapServer: 'broker1.eu-west-2.example.com:9092', // Or a pattern: /^broker1\./
    custom: ({ headers }) => {
      // Only a custom reaches the headers, the key or the partition
      return headers.source === 'warehouse-scanner'
    },
  },
  handler: onStockMoved,
})

Filter	Type	Description
`topic`	`FilterStringMatcher`	Matches the topic the record came from
`tombstone`	`boolean`	`true` takes only tombstones, `false` takes only records carrying a value. Leave it off to take both. See Tombstones
`eventSourceArn`	`FilterStringMatcher`	Matches the MSK cluster ARN on the event. A self-managed event carries no ARN, so nothing from one matches this
`bootstrapServer`	`FilterStringMatcher`	Matches if any one of the event's brokers matches. AWS sends them as a comma separated list and the router splits it for you
`custom`	`(input: KafkaFilterInput) => boolean \| Promise<boolean>`	Anything the other keys cannot express, given the decoded `headers` and `headerList`, the `topic`, the `tombstone` flag and the raw `record`. Can be async

FilterStringMatcher is string | RegExp | Array<string | RegExp>. See filters for how each form matches, including the * wildcard.

A redelivered batch carries no ARN and no brokers. Lambda sends a batch reported through batchItemFailures back as its records alone, which the AWS event documentation does not show. The mapping has already pinned those records to a cluster the function consumes, so eventSourceArn and bootstrapServer have nothing left to test and let the record through. A retried record therefore reaches the same route as its first attempt.

The record key, the partition and the headers are all reachable through custom and nowhere else. Headers arrive decoded, the partition is a plain number on record, and record.key is still the base64 AWS sent, or null on a record published without one. See Message headers for the shape headers takes.

custom gets no parsed value. The value is only decoded and validated once a route has matched, so route on the contents by decoding record.value yourself. See custom for where it sits in the filter order.

Handler

Handlers take one argument and return nothing.

import { logger } from '@lambda-event-router/base'
import type { KafkaRequest, KafkaResponse } from '@lambda-event-router/kafka'

export async function onStockMoved(
  request: KafkaRequest<StockMovement>,
): Promise<KafkaResponse> {
  const { sku, quantity } = request.value
  logger.info(`Moved ${quantity} of ${sku} at offset ${request.offset}`)
}

Request object

Field	Type	Description
`value`	`TValue`	The record value, base64 decoded and JSON parsed. A value that is not valid JSON reaches you as the decoded string, and a record carrying no value at all as `undefined`. An empty value arrives as `undefined` too, so use `tombstone` to tell a delete from an empty payload
`key`	`string \| undefined`	The record key, base64 decoded to text. Kafka partitions on this, so every record with the same key lands on the same partition. A producer can publish without a key, and then this is `undefined` and the broker spreads those records round robin
`topic`	`string`	The topic the record came from
`partition`	`number`	The partition within that topic
`offset`	`number`	Where the record sits in its partition. Unique per partition and increasing, so it makes a good idempotency key
`timestamp`	`number`	A Unix timestamp in milliseconds. `record.timestampType` says whether the producer or the broker set it
`headers`	`KafkaHeaders`	The record headers keyed by name, values decoded to text, and empty where the record carries none. A repeated name holds the last value sent. See Message headers
`headerList`	`KafkaDecodedHeader[]`	Every header entry in the order Kafka sent them, so a repeated name keeps both values
`tombstone`	`boolean`	Whether the record carries no value, which on a compacted topic means its key is deleted. See Tombstones
`record`	`KafkaRecord`	The untouched record from AWS, so `key` and `value` are still base64
`context`	`Context`	The Lambda context

record.key, record.value and record.headers can each be absent. All three are typed null | undefined, so narrow before you read them. The decoded key, value and headers on the request do that for you.

KafkaRecord, KafkaRecordHeader and KafkaDecodedHeader are declared by this package. Context comes from aws-lambda.

MSK and self-managed records are identical, so nothing in a handler has to know which cluster it is reading from. Only eventSourceArn differs, and that sits on the event rather than the record.

Response type

KafkaResponse is undefined. There is nothing useful to return from a Kafka record, so handlers return Promise<KafkaResponse> and the router works out what to hand back to Lambda.

Throwing is how you signal failure. See Failures and retries for what that does to the rest of the batch.

Inferred handlers

Nothing to look up and nothing to keep in sync. defineRoute reads the schema and hands your handler a fully typed value, defaults and coercion included, so quantity below is a number and damaged a boolean without you declaring either.

import { logger } from '@lambda-event-router/base'
import { defineRoute } from '@lambda-event-router/kafka'
import { z } from 'zod'

const StockMovementSchema = z.object({
  sku: z.string(),
  quantity: z.coerce.number(),
  damaged: z.coerce.boolean().default(false),
})

export const stockMovedRoute = defineRoute({
  filters: { topic: STOCK_TOPIC },
  valueSchema: StockMovementSchema,
}).handle(async ({ value, key, offset }) => {
  if (value.damaged) return
  logger.info(`Moved ${value.quantity} of ${value.sku} under key ${key} at offset ${offset}`)
})

kafkaRouter.route(stockMovedRoute)

Inference pays off most in a Lambda taking several event sources, since you never have to know any of their request shapes. See inferred handlers, where the same queue is written both ways to compare.

Annotated handlers

Annotating the request yourself splits route setup from business logic, using KafkaRequest and your own types.

// handlers/onStockMoved.ts
import { logger } from '@lambda-event-router/base'
import type { KafkaRequest, KafkaResponse } from '@lambda-event-router/kafka'
import { z } from 'zod'

export const StockMovementSchema = z.object({ sku: z.string(), quantity: z.number() })
type StockMovement = z.infer<typeof StockMovementSchema>

export async function onStockMoved(request: KafkaRequest<StockMovement>): Promise<KafkaResponse> {
  logger.info(`Moved ${request.value.quantity} of ${request.value.sku}`)
}

// kafka.ts
import { createKafkaRouter } from '@lambda-event-router/kafka'
import { onStockMoved, StockMovementSchema } from './handlers/onStockMoved'

const kafkaRouter = createKafkaRouter()

kafkaRouter.route({
  filters: { topic: STOCK_TOPIC },
  valueSchema: StockMovementSchema,
  handler: onStockMoved,
})

Derive the type from the schema with z.infer rather than hand-writing an interface that mirrors it. Exporting the schema from the handler file and attaching it to the route in the router file keeps the type and the validation from drifting apart. See annotated handlers for the worked version.

Schema validation

valueSchema is the only key that takes a schema, and it is optional.

const StockMovementSchema = z.object({
  sku: z.string(),
  quantity: z.coerce.number(),
  movedAt: z.coerce.date(),
})

kafkaRouter.route({
  filters: { topic: STOCK_TOPIC },
  valueSchema: StockMovementSchema,
  handler: onStockMoved,
})

Key	Validates
`valueSchema`	The decoded, parsed record value

Any Standard Schema library works. Validation runs after a route has matched, so a record failing its schema throws rather than falling through to the next route. See schema validation for what your handler receives after coercion.

An object schema fails every record whose value is not JSON. A producer writing Avro, Protobuf or plain text gets the decoded string through instead, which a z.object() rejects. Validate those with z.string() and decode in the handler, or leave the schema off.

A record carrying no value fails the same way, since the schema is handed undefined. Make the schema optional where that is something your producers send.

Nothing validates the key or the headers, so treat both as untrusted and the key as possibly absent.

Failures and retries

With batchItemFailures off, which is the default, records run one at a time in the order they arrive and the first throw fails the invocation. Lambda retries the whole batch, and both MaximumRetryAttempts and MaximumRecordAgeInSeconds default to infinite, so a record that always throws holds up everything behind it. Set those on the event source mapping, or give it an on-failure destination, to bound that. BisectBatchOnFunctionError splits a failing batch in two and retries each half, which narrows down which record is the problem.

Turn it on and the router reports the failure back to Lambda instead.

const kafkaRouter = createKafkaRouter({ batchItemFailures: true })

Reporting one failure reports every record after it in the same partition. A partition is ordered, so the router stops that partition at the first throw and returns the failing record along with every record left in it. Lambda redelivers that partition from the failed record onwards.

A batch can carry several partitions and the router handles each on its own, so a throw in one holds back only that partition. Records on a partition that succeeds end to end are neither reported nor redelivered.

Each failure Lambda gets back is an object rather than a plain id, pairing the topic-partition with the offset. KafkaBatchResponse is the shape the router hands back.

json

{
  "batchItemFailures": [
    { "itemIdentifier": { "partition": "stock-movements-0", "offset": 4218 } }
  ]
}

You also need to set the ReportBatchItemFailures response type on the event source mapping. Without it, AWS ignores what the router returns.

A redelivered batch is thinner than the first delivery. KafkaRetryEvent is that shape: records and nothing else. The router recognises it by its records, because there is no eventSource to name it by.

Middleware

Router and route middleware are both typed KafkaMiddleware, and the chain runs once per record, so a batch of ten records runs it ten times.

import { logger } from '@lambda-event-router/base'
import type { KafkaMiddleware } from '@lambda-event-router/kafka'

export const logInvocation: KafkaMiddleware = async (request, next) => {
  logger.info(`Handling ${request.topic}-${request.partition} at offset ${request.offset}`)
  return next(request)
}

export const withWarehouseContext: KafkaMiddleware<StockMovement> = async (request, next) => {
  logger.info(`Handling movement of ${request.value.sku}`)
  return next(request)
}

const kafkaRouter = createKafkaRouter({ middleware: [logInvocation] })

kafkaRouter.route({
  filters: { topic: STOCK_TOPIC },
  valueSchema: StockMovementSchema,
  middleware: [withWarehouseContext],
  handler: onStockMoved,
})

Route middleware carries the route's value type. A route with a valueSchema needs KafkaMiddleware<StockMovement>. KafkaMiddleware on its own does not compile there. Router middleware takes no type argument, because it runs for every route.

Records are handled one at a time, so appendKeys on the shared logger cannot interleave the way it does on SQS. Keys are cleared per invocation rather than per record, so one you set for a record stays on every record after it in the same batch. See middleware for the execution order and the three levels it attaches at.

Types

All exported from @lambda-event-router/kafka.

Type	Description
`KafkaRequest<TValue>`	The handler argument
`KafkaResponse`	Handler return type, `undefined`
`KafkaFilters`	The `filters` object
`KafkaFilterInput`	What `custom` receives
`KafkaRouteDefinition<TValue>`	A full route passed to `route()`
`KafkaRouterOptions`	Options for `createKafkaRouter`
`KafkaMiddleware<TValue>`	Router and route middleware
`KafkaHeaders`	Headers keyed by name, `Record<string, string>`
`KafkaDecodedHeader`	One entry of `headerList`, `Record<string, string>`
`KafkaBatchResponse`	What the router returns with `batchItemFailures` on
`KafkaBatchItemIdentifier`	One failed record, `{ partition, offset }`
`KafkaRecord`	One record as AWS sends it, with `key`, `value` and `headers` each able to be absent
`KafkaRecordHeader`	One raw header entry, `Record<string, number[]>`
`KafkaMSKEvent`, `KafkaSelfManagedEvent`	The two delivered event shapes
`KafkaRetryEvent`	A re-delivered batch, which carries `records` and nothing else
`KafkaEvent`	Any of the three event shapes

The KafkaRouter class and the createKafkaRouter and defineRoute functions come from the same place.

Generic parameters

The three types above that take a parameter take the same one.

Parameter	Types	Default
`TValue`	`request.value`	`unknown`

Leave it off and value is unknown, so KafkaRequest on its own is the right annotation for middleware that only reads the topic, the offset or the headers.

You only need this for annotated handlers. Inference covers it.

Tombstones

A tombstone is a record with a key and no value. It means that key is deleted.

Log compaction is why they exist. A topic set to cleanup.policy=compact keeps only the newest record per key. That makes it a table of key to latest value. Producing nothing would leave the old value in place, so you produce a null value instead.

Change-data-capture connectors send a tombstone for every deleted row.

tombstone on the request says whether a record is one. The filter key sends deletes to their own route.

kafkaRouter.route({
  filters: { topic: ORDERS_TOPIC, tombstone: true },
  handler: async ({ key }) => forgetOrder(key),
})

kafkaRouter.route({
  filters: { topic: ORDERS_TOPIC },
  valueSchema: OrderSchema,
  handler: upsertOrder,
})

Leave the filter off to take both, as the second route does. tombstone: false takes only records carrying a value.

A record with an empty value is not a tombstone. An empty payload is data, not a delete, so tombstone is false. Both reach the handler as value: undefined, so tombstone is what tells them apart.

Message headers

Kafka lets a producer attach any number of headers to a record, and it sends the values as bytes rather than text. AWS passes them through as a list, one object per header, whose single key is the header name, and leaves the field off a record with no headers at all.

The router decodes every value to UTF-8 and gives you both shapes. headers is keyed by header name, which is what you want almost every time.

kafkaRouter.route({
  filters: { topic: AUDIT_TOPIC },
  handler: async ({ headers }) => {
    if (headers.auditType === 'COMPLIANCE') await archive(headers.correlationId)
  },
})

A repeated header name collapses in headers, keeping the last value sent. Kafka allows a producer to send the same name twice. headerList keeps every entry in the order it arrived, so use that when a name can repeat.

const allTags = headerList.filter((header) => 'tag' in header).map((header) => header.tag)

Both are empty when the record carries no headers, so neither needs a guard.

Headers are the one piece of record metadata with no filter key of its own, so a custom is where a route picks on them.

kafkaRouter.route({
  filters: {
    topic: STOCK_TOPIC,
    custom: ({ headers }) => headers.source === 'warehouse-scanner',
  },
  handler: onScannedMovement,
})

Code example

A stock movements topic and a stock alerts topic on one MSK cluster, feeding a single Lambda.

Open a file: index.ts | Kafka router | handlers | schemas

TSkafka.ts

A record comes from one topic, so the two topic filters cannot both match and the order you register them in makes no difference. Point the event source mapping at a third topic without adding a route for it and every batch carrying one throws, so the two need to keep pace with each other.

onStockAlert returns early when there is plenty left. A custom cannot do that job here, since it never sees the parsed value.

index.ts hands the router to LambdaRouter, which is what AWS invokes and what every router in the Lambda gets registered on. See routers for how the two levels of matching fit together.

KafkaRouter ​

Install ​

Create the router ​

Options ​

Register routes ​

Filters ​

Handler ​

Request object ​

Response type ​

Inferred handlers ​

Annotated handlers ​

Schema validation ​

Failures and retries ​

Middleware ​

Types ​

Generic parameters ​

Tombstones ​

Message headers ​

Code example ​