Kafka share groups with Lambda: set up queue mode

Kafka share groups with Lambda solve a problem I keep running into in Kafka projects: concurrency is tied to the number of partitions. If a topic has six partitions, at most six consumers work on it at the same time. When a traffic spike hits, you either over-partition up front or reroute the messages into a second queue. Both are annoying. Moreover, both are things you have to operate forever.

Since November 2, 2026, AWS Lambda offers a queue mode for self-managed Kafka clusters (and services such as Confluent Cloud). It is built on the share groups from Apache Kafka 4.2 and decouples concurrency from partitions. In this article, therefore, we look at what is behind it, when the mode pays off and where it can hurt you. After that, you build a Lambda function with AWS CDK, step by step, that processes ride-ended events from a fictional bike rental. It does so idempotently, because in queue mode that is no longer optional.

In short

In queue mode, Lambda reads a Kafka topic through a share group: several pollers share the same partitions, each with up to ten concurrent invocations. However, you lose ordering, and duplicates become more likely. The biggest pitfall: without an on-failure destination, the broker drops a record after several failed deliveries.

What you will learn in this article

• How share groups differ from classic consumer groups
• When queue mode fits and when you should stay with stream mode
• How to try share groups locally with Docker in five minutes
• How to create the event source mapping in queue mode with AWS CDK
• What an idempotent TypeScript handler with DynamoDB looks like

What is new with Kafka share groups for Lambda?

Lambda reads Kafka through an event source mapping (ESM). That is the AWS-managed poller that fetches records, bundles them into batches and invokes your function with them. Previously, this ESM always used the classic consumer group protocol. The SelfManagedKafkaEventSourceConfig now has a ConsumptionMode field with two values:

• Stream (default): consumer group, each partition belongs to exactly one poller, ordering per partition is preserved.
• Queue: share group, all pollers read together from all partitions, without ordering but with more throughput.

One thing matters for planning: according to the AWS documentation, queue mode only works with self-managed Kafka clusters on version 4.2 or newer that have share groups switched on. Amazon MSK does not support it yet; AWS plans to follow once MSK offers Kafka 4.2 with share groups. Furthermore, you do not have to change producers, topic configuration or function code. What you do have to change is your assumptions about ordering and duplicates. More on that in a moment.

Why partitions used to cap concurrency

In a classic consumer group, Kafka assigns partitions exclusively to members. One member can own several partitions, but a partition never has several members. That is great when ordering matters, for example with account transactions. However, it gets in the way when every event is an independent task: resizing an image, generating an invoice, sending a push notification. As a result, a single slow record holds up everything behind it in the same partition.

Stream mode with fixed assignment on the left, queue mode with shared partitions on the right.

How share groups work in detail

Kafka share groups come from the improvement proposal KIP-932 (“Queues for Kafka”); Lambda requires a broker on version 4.2 or later. The broker keeps a state for every record. Initially, a free record is Available. When a consumer fetches it, it becomes Acquired and is locked for that one consumer for a limited time. Finally, once the consumer confirms processing, it becomes Acknowledged. If the lock expires or the consumer releases it on purpose, it is available again. Each fetch raises its delivery count by one.

Two broker settings are worth knowing. group.share.record.lock.duration.ms sets the lock duration; the default is 30 seconds. group.share.delivery.count.limit caps the delivery attempts; the default is 5. Once the limit is reached, the record moves to the Archived state and is not delivered again. For poison messages, that is what you want. However, for records that only failed because a mail server had a short hiccup, probably not.

🎬 Queue mode: records from one partition run in parallel on several Lambda invocations – this animated diagram only runs on my site: watch it on sascha-ms.de

For Kafka share groups, Lambda uses the share consumer with explicit acknowledgement. A poller fetches a batch, invokes your function and acknowledges only afterwards. Consequently, this leads to two restrictions: there is no batching window (MaximumBatchingWindowInSeconds) and no starting position (StartingPosition). Reading always resumes from the position the share group has already acknowledged.

Stream or queue: when does queue mode pay off?

Here are the limits from the AWS documentation side by side:

Lambda with Kafka share groups fits when every event stands on its own and speed matters more than ordering. Typical candidates are, for example, image and video processing, notifications, document generation or collecting telemetry. In addition, queue mode helps with topics that have few partitions and that you cannot rebuild because other teams depend on them.

Stay with stream mode as soon as a later event depends on an earlier one. For example, take “ride started” and “ride ended” for the same ride. In queue mode, the end can therefore arrive before the start. Similarly, counters and sums are tricky: if a record is processed twice, you count twice. AWS explicitly warns that unordered processing combined with duplicates can lead to wrong results.

Try Kafka share groups locally with Docker

Before you deploy anything to AWS, it is worth seeing the behavior for yourself. With the official Docker image, a Kafka broker runs in KRaft mode (without ZooKeeper) within seconds. From Kafka 4.2 on, share groups are enabled by default. Besides that, you only need Docker and three terminal windows.

Step 1: Start the broker and create a topic with just one partition. One partition is on purpose: this is exactly where a consumer group would let only a single consumer work.

docker run -d --name kafka -p 9092:9092 apache/kafka:4.4.0 docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic ride-ended --partitions 1 

Step 2: In two terminals, start one share consumer each in the same group.

docker exec -it kafka /opt/kafka/bin/kafka-console-share-consumer.sh \ --bootstrap-server localhost:9092 --topic ride-ended --group invoice-workers 

Step 3: In the third terminal, write a few lines to the topic. Each line is one record.

docker exec -it kafka /opt/kafka/bin/kafka-console-producer.sh \ --bootstrap-server localhost:9092 --topic ride-ended 

You will see both consumers splitting the lines between them, even though there is only one partition. In contrast, with a consumer group, one of them would sit idle. Finally, take a look at the group:

docker exec kafka /opt/kafka/bin/kafka-share-groups.sh --bootstrap-server localhost:9092 \ --describe --group invoice-workers --members 

The output lists both members with the same assignment ride-ended:0. That is the core idea of share groups. Clean up with docker rm -f kafka. If a command is named differently on your machine, --help shows the options of your version.

Step by step: Lambda with Kafka share groups in CDK

Now it gets concrete. The fictional bike rental writes a ride-ended event to Kafka after every ride. A Lambda function reads the topic through Kafka share groups and is supposed to generate and send an invoice from it. Ordering does not matter, but every ride must be billed only once. Therefore, it is clear what the code has to do: work in parallel and reliably detect duplicates.

Prerequisites

• A self-managed Kafka cluster on version 4.2 or later (or a provider such as Confluent Cloud) that Lambda can reach, here via SASL/SCRAM over public endpoints
• A secret in AWS Secrets Manager with username and password for Kafka access
• Node.js 22 or later, the AWS CLI and an AWS account with permissions for CDK deployments
• Basic knowledge of TypeScript and AWS CDK

Step 1: Create the project

mkdir ride-invoices && cd ride-invoices npm init -y npm install aws-cdk-lib constructs @aws-sdk/client-dynamodb @aws-sdk/lib-dynamodb npm install -D aws-cdk typescript tsx esbuild @types/node @types/aws-lambda mkdir bin lib src test 

Add a tsconfig.json and a cdk.json. I checked the code with TypeScript 7.0.2, aws-cdk-lib 2.275.0 and Node.js 22.

// tsconfig.json { "compilerOptions": { "target": "ES2022", "module": "NodeNext", "moduleResolution": "NodeNext", "strict": true, "esModuleInterop": true, "skipLibCheck": true, "noEmit": true, "types": ["node"] }, "include": ["bin", "lib", "src", "test"] } 

// cdk.json { "app": "npx tsx bin/app.ts" } 

The comment lines above the two JSON blocks are only there for orientation; leave them out of the real files, since JSON does not support comments.

Step 2: Understand and decode the Kafka event

The event format is the same in both modes. records is grouped by “topic-partition”, and key and value arrive base64-encoded. The only new thing is that in queue mode, records from several partitions land in one batch in any order. That is why the helper simply flattens the batch. It flags broken messages with a dedicated error type so that we can skip them on purpose later.

// src/kafka-event.ts // Minimal types for the Kafka event that Lambda delivers (Stream and Queue mode share the format). export interface KafkaHeader { [name: string]: number[]; } export interface KafkaRecord { topic: string; partition: number; offset: number; timestamp: number; timestampType: string; key?: string; // base64 value?: string; // base64 headers: KafkaHeader[]; } export interface KafkaEvent { eventSource: string; // "SelfManagedKafka" bootstrapServers?: string; records: Record<string, KafkaRecord[]>; // key: "<topic>-<partition>" } export interface RideEnded { rideId: string; bikeId: string; customerEmail: string; minutes: number; } // Flattens the batch. In Queue mode the order across and within partitions is not guaranteed. export function flattenRecords(event: KafkaEvent): KafkaRecord[] { return Object.values(event.records ?? {}).flat(); } export function decodeValue(record: KafkaRecord): RideEnded { if (!record.value) { throw new PoisonMessageError(`Record ${record.topic}/${record.partition}/${record.offset} has no value`); } let parsed: unknown; try { parsed = JSON.parse(Buffer.from(record.value, 'base64').toString('utf8')); } catch { throw new PoisonMessageError(`Record ${record.topic}/${record.partition}/${record.offset} is not valid JSON`); } const ride = parsed as Partial<RideEnded>; if (typeof ride.rideId !== 'string' || typeof ride.minutes !== 'number') { throw new PoisonMessageError(`Record ${record.topic}/${record.partition}/${record.offset} is missing rideId or minutes`); } return ride as RideEnded; } // Errors that will never succeed on retry (broken JSON, missing fields). export class PoisonMessageError extends Error { override name = 'PoisonMessageError'; } 

Step 3: Secure idempotency with DynamoDB

Idempotent means: no matter how often the same record arrives, the result is the same. The key for that is the business ID, here rideId, not the Kafka offset. Consequently, you also catch duplicates the producer created itself. I explained the pattern in detail in “Idempotency in AWS Lambda”. Here is a variant with a “claim”: an invocation reserves the ride with a conditional write, does the work and then marks it as done.

// src/idempotency.ts import { ConditionalCheckFailedException, DynamoDBClient } from '@aws-sdk/client-dynamodb'; import { DeleteCommand, DynamoDBDocumentClient, PutCommand, UpdateCommand } from '@aws-sdk/lib-dynamodb'; export type ClaimResult = 'claimed' | 'already-done' | 'in-progress'; // Small interface so the handler can be tested without AWS. export interface IdempotencyStore { claim(key: string, nowMs: number, lockMs: number): Promise<ClaimResult>; complete(key: string, nowMs: number): Promise<void>; release(key: string): Promise<void>; } const DAY_SECONDS = 24 * 60 * 60; export class DynamoIdempotencyStore implements IdempotencyStore { private readonly doc: DynamoDBDocumentClient; constructor(private readonly tableName: string, client = new DynamoDBClient({})) { this.doc = DynamoDBDocumentClient.from(client); } async claim(key: string, nowMs: number, lockMs: number): Promise<ClaimResult> { try { await this.doc.send( new PutCommand({ TableName: this.tableName, Item: { pk: key, status: 'IN_PROGRESS', lockedUntil: nowMs + lockMs, expiresAt: Math.floor(nowMs / 1000) + 7 * DAY_SECONDS, // TTL attribute }, // New key, or an abandoned claim whose lock has expired. ConditionExpression: 'attribute_not_exists(pk) OR (#s = :inProgress AND lockedUntil < :now)', ExpressionAttributeNames: { '#s': 'status' }, ExpressionAttributeValues: { ':inProgress': 'IN_PROGRESS', ':now': nowMs }, ReturnValuesOnConditionCheckFailure: 'ALL_OLD', }), ); return 'claimed'; } catch (err) { if (err instanceof ConditionalCheckFailedException) { return err.Item?.status?.S === 'DONE' ? 'already-done' : 'in-progress'; } throw err; } } async complete(key: string, nowMs: number): Promise<void> { await this.doc.send( new UpdateCommand({ TableName: this.tableName, Key: { pk: key }, UpdateExpression: 'SET #s = :done, doneAt = :now REMOVE lockedUntil', ExpressionAttributeNames: { '#s': 'status' }, ExpressionAttributeValues: { ':done': 'DONE', ':now': nowMs }, }), ); } async release(key: string): Promise<void> { await this.doc.send(new DeleteCommand({ TableName: this.tableName, Key: { pk: key } })); } } // In-memory variant for tests and local experiments. export class MemoryIdempotencyStore implements IdempotencyStore { readonly items = new Map<string, { status: 'IN_PROGRESS' | 'DONE'; lockedUntil: number }>(); async claim(key: string, nowMs: number, lockMs: number): Promise<ClaimResult> { const existing = this.items.get(key); if (existing?.status === 'DONE') return 'already-done'; if (existing && existing.lockedUntil >= nowMs) return 'in-progress'; this.items.set(key, { status: 'IN_PROGRESS', lockedUntil: nowMs + lockMs }); return 'claimed'; } async complete(key: string): Promise<void> { this.items.set(key, { status: 'DONE', lockedUntil: 0 }); } async release(key: string): Promise<void> { this.items.delete(key); } } 

The condition in the PutCommand is the most important part. It only allows the write if the key does not exist yet, or if an earlier claim was abandoned because its invocation crashed. On a conflict, ReturnValuesOnConditionCheckFailure returns the old item, so we know without a second read whether the ride is already done or currently being processed. Additionally, the TTL cleans up old items after seven days.

Step 4: Process the batch

Now for the actual logic. It knows three cases. A ride is new, so do the work. It is already done, so skip it. Or another invocation is holding it right now, in which case the batch should come back later. Poison messages such as broken JSON block nothing; they are logged and skipped.

// src/process-batch.ts import { decodeValue, flattenRecords, KafkaEvent, PoisonMessageError, RideEnded } from './kafka-event'; import { IdempotencyStore } from './idempotency'; export interface Dependencies { store: IdempotencyStore; sendInvoice: (ride: RideEnded) => Promise<void>; now?: () => number; log?: (msg: string, data?: Record<string, unknown>) => void; } export interface BatchSummary { processed: number; skippedDuplicates: number; poison: number; } // Longer than the function timeout: an abandoned claim only expires once the invocation is surely gone. export const CLAIM_LOCK_MS = 60_000; export class RetryLaterError extends Error { override name = 'RetryLaterError'; } export async function processBatch(event: KafkaEvent, deps: Dependencies): Promise<BatchSummary> { const now = deps.now ?? Date.now; const log = deps.log ?? ((msg, data) => console.log(JSON.stringify({ msg, ...data }))); const summary: BatchSummary = { processed: 0, skippedDuplicates: 0, poison: 0 }; let retryNeeded = false; for (const record of flattenRecords(event)) { let ride: RideEnded; try { ride = decodeValue(record); } catch (err) { if (err instanceof PoisonMessageError) { // Retrying will not help. Log it and move on so it does not block the batch. summary.poison++; log('poison message skipped', { reasonSource：https://dev.to/sascha-ms/kafka-share-groups-with-lambda-set-up-queue-mode-5462
My favorites

Related posts

Hexagonal Architecture for a System of AI Agents: model agnostic, one core, runs on PC and in the cloud

How I separated the rules every agent shares from the places those agents run, so that one core serves a fleet of agents

0dev.to AI

Your iPhone in Parallax, and agent taps 6 to 20 times faster

A Galaxy S24 and an iPhone 13 in one Parallax window, running the same test app on one Metro. The agent taps on the S24;

0dev.to AI

CSS Flexbox Order: How to Reorder Items (And the Accessibility Trap)

CSS Flexbox order is a double-edged sword. It visually moves elements around, but the DOM stays exactly the same. This b

0dev.to CSS

Test The Avatar Asset Before Adding A CSS Fix

An avatar with a white square may tempt you to add a background rule. First determine whether the square is inside the f

0dev.to CSS

HTML CSS débutant : le cours pas à pas, un mini-projet et 8 exercices corrigés

L'essentiel en une minute Ce cours HTML CSS débutant part de zéro et va jusqu'à une page publiée : structure, mise en fo

0dev.to CSS

Derive every margin from one module, a Swiss grid in code

I made one square control all the margins in a logo system. I would do it again. There is a trade off as nothing can be

0dev.to CSS