diff --git a/smk-lambda-queue-mode-python-sam/.gitignore b/smk-lambda-queue-mode-python-sam/.gitignore
new file mode 100644
index 000000000..20cfb4728
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/.gitignore
@@ -0,0 +1,5 @@
+.aws-sam/
+__pycache__/
+*.pyc
+.pytest_cache/
+samconfig.toml
diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md
new file mode 100644
index 000000000..529393152
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/README.md
@@ -0,0 +1,374 @@
+# Self-managed Apache Kafka Queue mode to AWS Lambda (KIP-932)
+
+This pattern deploys an AWS Lambda function that consumes from a self-managed Apache Kafka 4.2+ cluster using **Queue consumption mode** (KIP-932 Share Groups). Queue mode allows multiple Lambda pollers to process records from the same partition concurrently — parallelism is decoupled from partition count.
+
+Learn more about this pattern at Serverless Land: https://serverlessland.com/patterns/smk-lambda-queue-mode-python-sam
+
+> **Important:** Queue consumption mode requires Apache Kafka 4.2 or later with share groups enabled. This is a preview feature — `ConsumptionMode: Queue` is not yet available in the SAM or CloudFormation schema. The ESM is created via a script that calls the Lambda API directly.
+
+## Architecture
+
+```mermaid
+graph LR
+    Producer["Producer\nLambda"] -->|publish| Kafka["Apache Kafka\n4.2+ Cluster"]
+    Kafka -->|Queue mode ESM\nConsumptionMode: Queue| Worker["Worker\nLambda"]
+    Worker --> SQS["SQS\nDLQ"]
+    Worker --> CW["CloudWatch\nMetrics"]
+```
+
+**Queue mode vs Stream mode:** In Stream mode each partition maps to exactly one consumer — a 3-partition topic supports at most 3 concurrent Lambda invocations. In Queue mode, multiple pollers share all partitions — 10 pollers can process a 3-partition topic concurrently, and a slow record in one poller does not block other pollers.
+
+## Prerequisites
+
+- [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html) installed and configured
+- [AWS SAM CLI](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-sam-cli-install.html) installed
+- [Python 3.12](https://www.python.org/downloads/)
+- curl >= 7.75 (for `--aws-sigv4` support)
+- Apache Kafka 4.2+ cluster (see Deployment Path A to provision one automatically)
+
+## Costs
+
+This pattern uses EC2 (t3.medium), Lambda, SQS, CloudWatch, and VPC resources. See [AWS Pricing](https://aws.amazon.com/pricing/) for details. There are costs associated with these services beyond the Free Tier.
+
+---
+
+## Deployment
+
+Choose the path that matches your setup:
+
+| Path | When to use |
+|------|------------|
+| **A — Full** | Starting from scratch — provisions VPC, Kafka EC2 broker, and Lambda app |
+| **B — Bring your own Kafka** | You have an existing Kafka 4.2+ cluster; need VPC and Lambda app |
+| **C — Bring your own Kafka + VPC** | You have an existing Kafka 4.2+ cluster and VPC |
+
+### Clone the repository
+
+```bash
+git clone https://github.com/aws-samples/serverless-patterns
+cd serverless-patterns/smk-lambda-queue-mode-python-sam
+```
+
+---
+
+### Path A: Full deployment (VPC + Kafka broker + Lambda app)
+
+**Step 1: Deploy the network stack**
+
+```bash
+aws cloudformation deploy \
+  --stack-name kafka-queue-network \
+  --template-file stacks/1-network.yaml \
+  --region <region>
+```
+
+**Step 2: Deploy the Kafka broker**
+
+```bash
+aws cloudformation deploy \
+  --stack-name kafka-queue-broker \
+  --template-file stacks/2-broker.yaml \
+  --capabilities CAPABILITY_IAM \
+  --region <region>
+```
+
+This provisions a t3.medium EC2 instance. The instance is ready in ~2 minutes.
+
+**Step 2b: Install Kafka on the broker**
+
+```bash
+chmod +x scripts/setup-broker.sh
+./scripts/setup-broker.sh --region <region> --profile <profile>
+```
+
+This script connects to the broker via SSM (no SSH required) and installs Apache Kafka 4.2.x, configures KRaft mode with share groups enabled, and starts the broker. It runs 6 steps sequentially and reports progress. The Kafka download (~130MB) takes about 15-20 minutes depending on network speed.
+
+**Step 3: Build and deploy the application**
+
+```bash
+sam build --template stacks/3-app.yaml
+sam deploy \
+  --stack-name kafka-queue-app \
+  --template-file .aws-sam/build/template.yaml \
+  --capabilities CAPABILITY_IAM \
+  --resolve-s3 \
+  --region <region>
+```
+
+**Step 4: Create VPC endpoints**
+
+ESM pollers run in private subnets and require three VPC interface endpoints: `lambda`, `sts`, and `sqs`. Missing any one of them — especially `sqs` — causes a silent failure where the ESM stays `Enabled/OK` but Lambda stops being invoked after the first batch.
+
+```bash
+chmod +x scripts/setup-vpc-endpoints.sh
+./scripts/setup-vpc-endpoints.sh --region <region> --profile <profile>
+```
+
+This script is idempotent — it skips endpoints that already exist.
+
+**Step 5: Create the Queue mode ESM**
+
+```bash
+chmod +x scripts/create-esm.sh
+./scripts/create-esm.sh --region <region> --profile <profile>
+```
+
+Note the ESM UUID printed in the output — you need it for the observability step.
+
+Wait ~60 seconds for the ESM to reach `State: Enabled`.
+
+**Step 6: Deploy observability (optional)**
+
+```bash
+aws cloudformation deploy \
+  --stack-name kafka-queue-observability \
+  --template-file stacks/4-observability.yaml \
+  --parameter-overrides ESMUuid=<esm-uuid-from-step-5> \
+  --region <region>
+```
+
+This creates a CloudWatch dashboard (`kafka-queue-dashboard`) and alarms for share group lag, DLQ delivery, and poller errors.
+
+---
+
+### Path B: Bring your own Kafka cluster
+
+Skip Steps 1, 2, and 2b. Provide your Kafka bootstrap servers, VPC subnet IDs, and security group at deploy time:
+
+```bash
+sam build --template stacks/3-app.yaml
+sam deploy \
+  --stack-name kafka-queue-app \
+  --template-file .aws-sam/build/template.yaml \
+  --capabilities CAPABILITY_IAM \
+  --resolve-s3 \
+  --parameter-overrides \
+    UseExistingInfra=true \
+    BootstrapServers="broker1.example.com:9092,broker2.example.com:9092" \
+    VpcSubnetIds="subnet-aaa111,subnet-bbb222,subnet-ccc333" \
+    VpcSecurityGroupId="sg-eee555" \
+  --region <region>
+```
+
+Your Kafka cluster must:
+- Run Apache Kafka 4.2 or later
+- Have `group.coordinator.rebalance.protocols=classic,consumer,share` in `server.properties`
+- Have `share.version` upgraded to 1 via `kafka-features.sh upgrade --feature share.version=1`
+- Be reachable from the Lambda VPC subnets on the configured port
+
+**Create VPC endpoints**
+
+If your VPC does not already have `lambda`, `sts`, and `sqs` interface endpoints, create them:
+
+```bash
+./scripts/setup-vpc-endpoints.sh --region <region> --profile <profile> \
+  --vpc-id <your-vpc-id> \
+  --subnet-ids "subnet-aaa111,subnet-bbb222,subnet-ccc333" \
+  --security-group-id <your-security-group-id>
+```
+
+If your VPC already has these endpoints, skip this step.
+
+**Create the Queue mode ESM**
+
+```bash
+./scripts/create-esm.sh --region <region> --profile <profile>
+```
+
+---
+
+### Path C: Bring your own Kafka cluster and VPC
+
+Same as Path B — `UseExistingInfra=true` handles both cases. Provide your own `--vpc-id`, `--subnet-ids`, and `--security-group-id` to `setup-vpc-endpoints.sh` if your VPC doesn't already have the required endpoints.
+
+---
+
+---
+
+## Testing
+
+**Produce 20 records:**
+
+```bash
+aws lambda invoke \
+  --function-name kafka-queue-app-producer \
+  --region <region> \
+  --cli-binary-format raw-in-base64-out \
+  --payload '{"count": 20}' /dev/stdout
+```
+
+You can also override the topic at invocation time without redeploying:
+
+```bash
+aws lambda invoke \
+  --function-name kafka-queue-app-producer \
+  --region <region> \
+  --cli-binary-format raw-in-base64-out \
+  --payload '{"count": 20, "topic": "my-custom-topic"}' /dev/stdout
+```
+
+Every 7th record (`taskIndex % 7 == 0`) has `shouldFail: true` to demonstrate the RELEASE/retry/DLQ path.
+
+**Watch the worker Lambda logs:**
+
+```bash
+aws logs tail /aws/lambda/kafka-queue-app-worker \
+  --follow \
+  --filter-pattern KAFKA_RECORD \
+  --region <region>
+```
+
+You should see `KAFKA_RECORD` log entries with `topic`, `partition`, `offset`, and `payload`. Records with `shouldFail: true` log a warning and return in `batchItemFailures`, causing the broker to RELEASE them for retry.
+
+**View the CloudWatch dashboard:**
+
+Open the `kafka-queue-dashboard` dashboard in CloudWatch to see real-time metrics:
+- **Share Group Lag** — records waiting to be processed (spikes on produce, drains to zero)
+- **Provisioned Pollers** — number of active pollers (Queue mode only)
+- **Event Counts** — PolledEventCount, InvokedEventCount, OnFailureDestinationDeliveredEventCount
+- **Errors** — PollingErrorCount, FailedInvokeEventCount
+
+All metrics are scoped to the ESM UUID, not the function name.
+
+**Test locally:**
+
+```bash
+sam local invoke WorkerFunction \
+  --template stacks/3-app.yaml \
+  --event events/kafka-event.json
+```
+
+---
+
+## Verifying Queue mode scaling
+
+The key differentiator of Queue mode is that pollers exceed the partition count. Verify this directly on the broker after producing a large batch:
+
+**Step 1: Produce a large batch**
+
+```bash
+aws lambda invoke \
+  --function-name kafka-queue-app-producer \
+  --region <region> \
+  --cli-binary-format raw-in-base64-out \
+  --payload '{"count": 200}' /dev/stdout
+```
+
+**Step 2: Check the broker coordinator log**
+
+Connect to the broker via SSM Session Manager and run:
+
+```bash
+grep "new assignment state" /var/log/kafka.log | grep <your-consumer-group-id> | tail -20
+```
+
+You should see multiple members assigned to the same partition simultaneously. For example, with a 3-partition topic and `MaximumPollers: 10`, the output shows more than 3 members total — some partitions shared by 2 or more pollers:
+
+```
+[GroupId my-queue-group] Member AAA new assignment state: ... assignedPartitions=[topic-0]
+[GroupId my-queue-group] Member BBB new assignment state: ... assignedPartitions=[topic-0]  <- same partition!
+[GroupId my-queue-group] Member CCC new assignment state: ... assignedPartitions=[topic-1]
+[GroupId my-queue-group] Member DDD new assignment state: ... assignedPartitions=[topic-1]  <- same partition!
+[GroupId my-queue-group] Member EEE new assignment state: ... assignedPartitions=[topic-2]
+```
+
+In Stream mode, each partition can only appear once across all members. Multiple members sharing the same partition is only possible with share groups — this is the Queue mode scaling proof.
+
+**Step 3: Confirm via kafka-share-groups.sh**
+
+```bash
+# On the broker
+bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --list
+# Your consumer group ID should appear here (not in kafka-consumer-groups.sh)
+
+bin/kafka-share-groups.sh --bootstrap-server localhost:9092 \
+  --describe --group <your-consumer-group-id>
+# Shows per-partition lag — confirms share group is consuming
+```
+
+If the group appears in `kafka-share-groups.sh` but NOT in `kafka-consumer-groups.sh`, it is a share group and Queue mode is active.
+
+---
+
+## Broker-side prerequisites
+
+Before Queue mode can work, ensure your Kafka 4.2+ broker has:
+
+```properties
+# Required: enables the share rebalance protocol
+group.coordinator.rebalance.protocols=classic,consumer,share
+
+# Required for single-broker setups (default is 3)
+share.coordinator.state.topic.replication.factor=1
+share.coordinator.state.topic.min.isr=1
+
+# Queue mode tuning parameters
+group.share.partition.max.record.locks=100
+group.share.record.lock.duration.ms=30000
+group.share.delivery.count.limit=3
+group.share.max.size=10
+```
+
+And run after broker start:
+
+```bash
+bin/kafka-features.sh --bootstrap-server localhost:9092 \
+  upgrade --feature share.version=1
+```
+
+---
+
+## Cleanup
+
+Delete stacks in reverse order:
+
+```bash
+# 1. Delete the ESM first (get UUID from create-esm.sh output or console)
+aws lambda delete-event-source-mapping --uuid <esm-uuid> --region <region>
+
+# 2. Get your VPC ID
+aws cloudformation describe-stacks \
+  --stack-name kafka-queue-network \
+  --query 'Stacks[0].Outputs[?OutputKey==`VpcId`].OutputValue' \
+  --output text --region <region>
+
+# 3. Find and delete the lambda VPC endpoint
+aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=<vpc-id>" "Name=service-name,Values=com.amazonaws.<region>.lambda" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region <region>
+
+aws ec2 delete-vpc-endpoints --vpc-endpoint-ids <lambda-endpoint-id> --region <region>
+
+# 4. Find and delete the sts VPC endpoint
+aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=<vpc-id>" "Name=service-name,Values=com.amazonaws.<region>.sts" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region <region>
+
+aws ec2 delete-vpc-endpoints --vpc-endpoint-ids <sts-endpoint-id> --region <region>
+
+# 5. Find and delete the sqs VPC endpoint
+aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=<vpc-id>" "Name=service-name,Values=com.amazonaws.<region>.sqs" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region <region>
+
+aws ec2 delete-vpc-endpoints --vpc-endpoint-ids <sqs-endpoint-id> --region <region>
+
+# 6. Delete application stacks
+aws cloudformation delete-stack --stack-name kafka-queue-observability --region <region>
+
+aws cloudformation delete-stack --stack-name kafka-queue-app --region <region>
+
+# 7. Delete broker (if deployed)
+aws cloudformation delete-stack --stack-name kafka-queue-broker --region <region>
+
+# 8. Delete network (if deployed)
+aws cloudformation delete-stack --stack-name kafka-queue-network --region <region>
+```
+
+---
+
+## Pattern details
+
+| Property | Value |
+|----------|-------|
+| Kafka version required | Apache Kafka 4.2+ |
+| Lambda runtime | Python 3.12 |
+| IaC framework | AWS SAM + AWS CloudFormation |
+| Delivery semantics | At-least-once |
+| Ordering guarantees | None (Queue mode) |
+| Authentication | PLAINTEXT (see notes for SASL/SCRAM) |
+| Region | Configurable |
diff --git a/smk-lambda-queue-mode-python-sam/events/kafka-event.json b/smk-lambda-queue-mode-python-sam/events/kafka-event.json
new file mode 100644
index 000000000..4b38bd8f7
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/events/kafka-event.json
@@ -0,0 +1,40 @@
+{
+  "eventSource": "aws:kafka",
+  "bootstrapServers": "10.0.0.43:9092",
+  "records": {
+    "kafka-queue-task-worker-0": [
+      {
+        "topic": "kafka-queue-task-worker",
+        "partition": 0,
+        "offset": 0,
+        "timestamp": 1726000000000,
+        "timestampType": "CREATE_TIME",
+        "key": "am9iSWQtMQ==",
+        "value": "eyJqb2JJZCI6ICJqb2JJZC0xIiwgInRhc2tJbmRleCI6IDEsICJwYXlsb2FkIjogInRhc2stMSIsICJzaG91bGRGYWlsIjogZmFsc2V9",
+        "headers": []
+      },
+      {
+        "topic": "kafka-queue-task-worker",
+        "partition": 0,
+        "offset": 1,
+        "timestamp": 1726000001000,
+        "timestampType": "CREATE_TIME",
+        "key": "am9iSWQtMg==",
+        "value": "eyJqb2JJZCI6ICJqb2JJZC0yIiwgInRhc2tJbmRleCI6IDcsICJwYXlsb2FkIjogInRhc2stNyIsICJzaG91bGRGYWlsIjogdHJ1ZX0=",
+        "headers": []
+      }
+    ],
+    "kafka-queue-task-worker-1": [
+      {
+        "topic": "kafka-queue-task-worker",
+        "partition": 1,
+        "offset": 0,
+        "timestamp": 1726000000500,
+        "timestampType": "CREATE_TIME",
+        "key": "am9iSWQtMw==",
+        "value": "eyJqb2JJZCI6ICJqb2JJZC0zIiwgInRhc2tJbmRleCI6IDIsICJwYXlsb2FkIjogInRhc2stMiIsICJzaG91bGRGYWlsIjogZmFsc2V9",
+        "headers": []
+      }
+    ]
+  }
+}
diff --git a/smk-lambda-queue-mode-python-sam/example-pattern.json b/smk-lambda-queue-mode-python-sam/example-pattern.json
new file mode 100644
index 000000000..ee88d2bc4
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/example-pattern.json
@@ -0,0 +1,70 @@
+{
+  "title": "Self-managed Apache Kafka Queue mode to AWS Lambda (KIP-932)",
+  "description": "Deploy a Lambda function that consumes from a self-managed Apache Kafka 4.2+ cluster using Queue consumption mode (KIP-932 Share Groups). Queue mode allows multiple Lambda pollers to process records from the same partition concurrently, breaking the partition-count ceiling of traditional consumer groups.",
+  "language": "Python",
+  "level": "300",
+  "framework": "AWS SAM",
+  "services": {
+    "from": "kafka",
+    "to": "lambda"
+  },
+  "introBox": {
+    "headline": "How it works",
+    "text": [
+      "With Queue consumption mode (ConsumptionMode: Queue), Lambda event pollers use Kafka Share Groups (KIP-932) instead of traditional consumer groups. Multiple pollers can read from the same partition concurrently — parallelism is decoupled from partition count.",
+      "This pattern deploys: a dedicated VPC with private subnets and VPC endpoints, a self-managed Kafka 4.2+ broker on EC2 with share groups enabled, a Python Lambda worker function with partial batch response, a producer Lambda for testing, and CloudWatch observability.",
+      "Deployment is split into four independent stacks so you can skip the network and broker stacks if you have existing infrastructure."
+    ]
+  },
+  "gitHub": {
+    "template": {
+      "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/smk-lambda-queue-mode-python-sam",
+      "templateURL": "serverless-patterns/smk-lambda-queue-mode-python-sam",
+      "projectFolder": "smk-lambda-queue-mode-python-sam",
+      "templateFile": "stacks/3-app.yaml"
+    }
+  },
+  "resources": {
+    "bullets": [
+      {
+        "text": "Using Lambda with self-managed Apache Kafka",
+        "link": "https://docs.aws.amazon.com/lambda/latest/dg/with-kafka.html"
+      },
+      {
+        "text": "Apache Kafka event poller scaling modes in Lambda",
+        "link": "https://docs.aws.amazon.com/lambda/latest/dg/kafka-scaling-modes.html"
+      },
+      {
+        "text": "Configuring error handling controls for Kafka event sources",
+        "link": "https://docs.aws.amazon.com/lambda/latest/dg/kafka-retry-configurations.html"
+      }
+    ]
+  },
+  "deploy": {
+    "text": [
+      "See README.md for full deployment instructions including three paths: full deployment, bring-your-own Kafka, and bring-your-own Kafka and VPC."
+    ]
+  },
+  "testing": {
+    "text": [
+      "See the GitHub repo for detailed testing instructions."
+    ]
+  },
+  "cleanup": {
+    "text": [
+      "Delete the ESM first: <code>aws lambda delete-event-source-mapping --uuid &lt;esm-uuid&gt;</code>, then delete stacks in reverse order: <code>kafka-queue-observability</code>, <code>kafka-queue-app</code>, <code>kafka-queue-broker</code>, <code>kafka-queue-network</code>."
+    ]
+  },
+  "authors": [
+    {
+      "name": "Vaibhav Jain",
+      "bio": "AWS - Senior Delivery Consultant. Specializes in data platform modernization and large-scale distributed systems.",
+      "linkedin": "https://www.linkedin.com/in/vaibhavjainv/"
+    },
+    {
+      "name": "Hardith Murari",
+      "bio": "AWS - Delivery Consultant. Specializes in Application Architecture and Modernization.",
+      "linkedin": "https://www.linkedin.com/in/hardithsuvarna/"
+    }
+  ]
+}
diff --git a/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh b/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh
new file mode 100755
index 000000000..20fdd1396
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh
@@ -0,0 +1,139 @@
+#!/usr/bin/env bash
+# create-esm.sh — Create a Kafka Queue mode ESM via direct API call
+#
+# ConsumptionMode: Queue is a pre-release field not yet in the CloudFormation
+# or SAM schema. This script creates the ESM by calling the Lambda REST API
+# directly using curl with AWS SigV4 signing, bypassing client-side validation.
+#
+# Prerequisites:
+#   - aws CLI configured with appropriate credentials
+#   - curl >= 7.75 (for --aws-sigv4 support)
+#   - Stacks kafka-queue-network, kafka-queue-broker, kafka-queue-app must be deployed
+#
+# Usage:
+#   ./scripts/create-esm.sh
+#   ./scripts/create-esm.sh --region us-west-2 --profile myprofile
+
+set -euo pipefail
+
+# ── Defaults ─────────────────────────────────────────────────
+REGION="${AWS_DEFAULT_REGION:-us-east-1}"
+PROFILE="${AWS_PROFILE:-default}"
+APP_STACK="kafka-queue-app"
+BROKER_STACK="kafka-queue-broker"
+NETWORK_STACK="kafka-queue-network"
+CONSUMER_GROUP_ID="kafka-queue-group-$(date +%s)"
+TOPIC="kafka-queue-task-worker"
+MIN_POLLERS=2
+MAX_POLLERS=10
+MAX_RETRY_ATTEMPTS=3
+
+# Parse optional flags
+while [[ $# -gt 0 ]]; do
+  case $1 in
+    --region)     REGION="$2";  shift 2 ;;
+    --profile)    PROFILE="$2"; shift 2 ;;
+    --topic)      TOPIC="$2";   shift 2 ;;
+    --group)      CONSUMER_GROUP_ID="$2"; shift 2 ;;
+    *) echo "Unknown option: $1"; exit 1 ;;
+  esac
+done
+
+echo "=== Creating Kafka Queue mode ESM ==="
+echo "  Region:        $REGION"
+echo "  Profile:       $PROFILE"
+echo "  Topic:         $TOPIC"
+echo "  Consumer Group: $CONSUMER_GROUP_ID"
+echo ""
+
+# ── Resolve values from stack outputs ────────────────────────
+get_output() {
+  aws cloudformation describe-stacks \
+    --stack-name "$1" --profile "$PROFILE" --region "$REGION" \
+    --query "Stacks[0].Outputs[?OutputKey=='$2'].OutputValue" \
+    --output text
+}
+
+echo "Fetching stack outputs..."
+WORKER_ARN=$(get_output "$APP_STACK" "WorkerFunctionArn")
+DLQ_ARN=$(get_output "$APP_STACK" "TaskWorkerDLQArn")
+BOOTSTRAP=$(get_output "$BROKER_STACK" "BootstrapServers")
+SUBNET_A=$(get_output "$NETWORK_STACK" "PrivateSubnetA")
+SUBNET_B=$(get_output "$NETWORK_STACK" "PrivateSubnetB")
+SUBNET_C=$(get_output "$NETWORK_STACK" "PrivateSubnetC")
+LAMBDA_SG=$(get_output "$NETWORK_STACK" "LambdaSecurityGroupId")
+
+echo "  Worker ARN:    $WORKER_ARN"
+echo "  Bootstrap:     $BOOTSTRAP"
+echo "  DLQ ARN:       $DLQ_ARN"
+echo ""
+
+# ── Export credentials for curl --aws-sigv4 ──────────────────
+echo "Exporting credentials for SigV4 signing..."
+unset AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN
+eval "$(aws configure export-credentials --format env --profile "$PROFILE")"
+
+# ── Create ESM via Lambda REST API ───────────────────────────
+echo "Creating ESM..."
+RESPONSE=$(curl -sS -X POST \
+  "https://lambda.$REGION.amazonaws.com/2015-03-31/event-source-mappings/" \
+  --aws-sigv4 "aws:amz:$REGION:lambda" \
+  --user "$AWS_ACCESS_KEY_ID:$AWS_SECRET_ACCESS_KEY" \
+  -H "x-amz-security-token: $AWS_SESSION_TOKEN" \
+  -H "Content-Type: application/json" \
+  -d "{
+    \"FunctionName\": \"$WORKER_ARN\",
+    \"SelfManagedEventSource\": {
+      \"Endpoints\": {\"KAFKA_BOOTSTRAP_SERVERS\": [\"$BOOTSTRAP\"]}
+    },
+    \"SelfManagedKafkaEventSourceConfig\": {
+      \"ConsumerGroupId\": \"$CONSUMER_GROUP_ID\",
+      \"ConsumptionMode\": \"Queue\"
+    },
+    \"Topics\": [\"$TOPIC\"],
+    \"BatchSize\": 10,
+    \"MaximumRetryAttempts\": $MAX_RETRY_ATTEMPTS,
+    \"FunctionResponseTypes\": [\"ReportBatchItemFailures\"],
+    \"DestinationConfig\": {
+      \"OnFailure\": {\"Destination\": \"$DLQ_ARN\"}
+    },
+    \"ProvisionedPollerConfig\": {
+      \"MinimumPollers\": $MIN_POLLERS,
+      \"MaximumPollers\": $MAX_POLLERS
+    },
+    \"MetricsConfig\": {
+      \"Metrics\": [\"EventCount\", \"ErrorCount\", \"KafkaMetrics\"]
+    },
+    \"SourceAccessConfigurations\": [
+      {\"Type\": \"VPC_SUBNET\", \"URI\": \"subnet:$SUBNET_A\"},
+      {\"Type\": \"VPC_SUBNET\", \"URI\": \"subnet:$SUBNET_B\"},
+      {\"Type\": \"VPC_SUBNET\", \"URI\": \"subnet:$SUBNET_C\"},
+      {\"Type\": \"VPC_SECURITY_GROUP\", \"URI\": \"security_group:$LAMBDA_SG\"}
+    ]
+  }")
+
+echo "$RESPONSE" | python3 -c "
+import sys, json
+r = json.load(sys.stdin)
+if 'Type' in r and r.get('Type') == 'User':
+    print('ERROR:', r.get('message'))
+    sys.exit(1)
+print('ESM created!')
+print('  UUID:            ', r.get('UUID'))
+print('  State:           ', r.get('State'))
+print('  ConsumptionMode: ', r.get('SelfManagedKafkaEventSourceConfig',{}).get('ConsumptionMode'))
+print('  ProvisionedPollers:', r.get('ProvisionedPollerConfig'))
+print()
+print('Wait ~60s for State to reach Enabled, then produce records:')
+print()
+print('  aws lambda invoke \\\\')
+print('    --function-name kafka-queue-app-producer \\\\')
+print('    --region $REGION --profile $PROFILE \\\\')
+print('    --cli-binary-format raw-in-base64-out \\\\')
+print('    --payload \'{"count": 20}\' /dev/stdout')
+print()
+print('Watch logs:')
+print('  aws logs tail /aws/lambda/kafka-queue-app-worker --follow \\\\')
+print('    --filter-pattern KAFKA_RECORD \\\\')
+print('    --region $REGION --profile $PROFILE')
+" REGION="$REGION" PROFILE="$PROFILE"
diff --git a/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh b/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh
new file mode 100755
index 000000000..7ef6d3bba
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh
@@ -0,0 +1,153 @@
+#!/usr/bin/env bash
+# setup-broker.sh — Install and start Kafka on the broker EC2 instance via SSM
+#
+# This script installs Apache Kafka 4.2.x on the EC2 instance deployed by
+# stacks/2-broker.yaml. It uses SSM Run Command so no SSH or bastion host
+# is required. Each step runs sequentially and reports progress.
+#
+# Usage:
+#   ./scripts/setup-broker.sh --region <region> --profile <profile>
+#
+# Prerequisites:
+#   - stacks/1-network.yaml and stacks/2-broker.yaml must be deployed
+#   - AWS CLI configured with appropriate credentials
+
+set -euo pipefail
+
+REGION="${AWS_DEFAULT_REGION:-us-east-1}"
+PROFILE="${AWS_PROFILE:-default}"
+BROKER_STACK="kafka-queue-broker"
+KAFKA_VERSION="4.2.0"
+
+while [[ $# -gt 0 ]]; do
+  case $1 in
+    --region)  REGION="$2";  shift 2 ;;
+    --profile) PROFILE="$2"; shift 2 ;;
+    *) echo "Unknown option: $1"; exit 1 ;;
+  esac
+done
+
+echo "=== Kafka broker setup ==="
+echo "  Region:  $REGION"
+echo "  Profile: $PROFILE"
+echo "  Version: $KAFKA_VERSION"
+echo ""
+
+# ── Get instance ID from stack output ────────────────────────
+INSTANCE_ID=$(aws cloudformation describe-stacks \
+  --stack-name "$BROKER_STACK" --profile "$PROFILE" --region "$REGION" \
+  --query "Stacks[0].Outputs[?OutputKey=='BrokerInstanceId'].OutputValue" \
+  --output text)
+
+if [ -z "$INSTANCE_ID" ] || [ "$INSTANCE_ID" = "None" ]; then
+  echo "ERROR: Could not get instance ID from stack $BROKER_STACK"
+  exit 1
+fi
+echo "Instance: $INSTANCE_ID"
+echo ""
+
+# ── Wait for SSM agent ────────────────────────────────────────
+echo "Waiting for SSM agent..."
+for i in $(seq 1 30); do
+  STATUS=$(aws ssm describe-instance-information \
+    --filters "Key=InstanceIds,Values=$INSTANCE_ID" \
+    --profile "$PROFILE" --region "$REGION" \
+    --query 'InstanceInformationList[0].PingStatus' \
+    --output text 2>/dev/null || echo "None")
+  if [ "$STATUS" = "Online" ]; then
+    echo "SSM agent ready."
+    break
+  fi
+  echo "  [$i/30] Waiting... ($STATUS)"
+  sleep 10
+done
+
+# ── Helper: run SSM command and wait ─────────────────────────
+run_ssm() {
+  local STEP="$1"
+  local CMD="$2"
+  local TIMEOUT="${3:-300}"
+
+  echo "[$STEP]"
+  CMD_ID=$(aws ssm send-command \
+    --instance-ids "$INSTANCE_ID" \
+    --document-name AWS-RunShellScript \
+    --parameters "commands=[\"$CMD\"]" \
+    --profile "$PROFILE" --region "$REGION" \
+    --query 'Command.CommandId' --output text)
+
+  # Poll until complete
+  for i in $(seq 1 $((TIMEOUT / 10))); do
+    sleep 10
+    STATUS=$(aws ssm get-command-invocation \
+      --command-id "$CMD_ID" --instance-id "$INSTANCE_ID" \
+      --profile "$PROFILE" --region "$REGION" \
+      --query 'Status' --output text 2>/dev/null || echo "Pending")
+    if [ "$STATUS" = "Success" ]; then
+      OUT=$(aws ssm get-command-invocation \
+        --command-id "$CMD_ID" --instance-id "$INSTANCE_ID" \
+        --profile "$PROFILE" --region "$REGION" \
+        --query 'StandardOutputContent' --output text 2>/dev/null)
+      [ -n "$OUT" ] && echo "  $OUT"
+      echo "  Done."
+      return 0
+    elif [ "$STATUS" = "Failed" ] || [ "$STATUS" = "TimedOut" ]; then
+      ERR=$(aws ssm get-command-invocation \
+        --command-id "$CMD_ID" --instance-id "$INSTANCE_ID" \
+        --profile "$PROFILE" --region "$REGION" \
+        --query 'StandardErrorContent' --output text 2>/dev/null)
+      echo "  ERROR: $ERR"
+      exit 1
+    fi
+    echo "  [$((i * 10))s] $STATUS..."
+  done
+  echo "  ERROR: Timed out after ${TIMEOUT}s"
+  exit 1
+}
+
+# ── Step 1: Install Java ──────────────────────────────────────
+run_ssm "1/6 Install Java" \
+  "dnf install -y java-21-amazon-corretto-headless && java -version 2>&1 | head -1" \
+  180
+
+# ── Step 2: Download Kafka ────────────────────────────────────
+run_ssm "2/6 Download Kafka $KAFKA_VERSION (this takes ~20 min)" \
+  "wget -q --timeout=2400 --tries=3 https://archive.apache.org/dist/kafka/${KAFKA_VERSION}/kafka_2.13-${KAFKA_VERSION}.tgz -O /tmp/kafka.tgz && ls -lh /tmp/kafka.tgz" \
+  2400
+
+# ── Step 3: Extract ───────────────────────────────────────────
+run_ssm "3/6 Extract Kafka" \
+  "mkdir -p /opt/kafka && tar -xzf /tmp/kafka.tgz -C /opt/kafka --strip-components=1 && echo extracted" \
+  120
+
+# ── Step 4: Write config and format storage ───────────────────
+run_ssm "4/6 Configure and format storage" \
+  "INSTANCE_IP=\$(curl -s http://169.254.169.254/latest/meta-data/local-ipv4) && mkdir -p /var/kafka-logs && printf '%s\n' process.roles=broker,controller node.id=1 controller.quorum.voters=1@localhost:9093 listeners=CONTROLLER://localhost:9093,PLAINTEXT://0.0.0.0:9092 advertised.listeners=PLAINTEXT://\$INSTANCE_IP:9092 listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT controller.listener.names=CONTROLLER inter.broker.listener.name=PLAINTEXT log.dirs=/var/kafka-logs num.partitions=3 default.replication.factor=1 offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 auto.create.topics.enable=true group.coordinator.rebalance.protocols=classic,consumer,share share.coordinator.state.topic.replication.factor=1 share.coordinator.state.topic.min.isr=1 group.share.partition.max.record.locks=100 group.share.record.lock.duration.ms=30000 group.share.delivery.count.limit=3 group.share.max.size=10 > /tmp/kraft-server.properties && CLUSTER_ID=\$(/opt/kafka/bin/kafka-storage.sh random-uuid) && /opt/kafka/bin/kafka-storage.sh format -t \$CLUSTER_ID -c /tmp/kraft-server.properties && echo formatted" \
+  60
+
+# ── Step 5: Start Kafka and wait for ready ────────────────────
+run_ssm "5/6 Start Kafka broker" \
+  "export KAFKA_HEAP_OPTS='-Xmx512m -Xms256m' && nohup /opt/kafka/bin/kafka-server-start.sh /tmp/kraft-server.properties > /var/log/kafka.log 2>&1 & sleep 20 && for i in \$(seq 1 12); do /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 > /dev/null 2>&1 && echo broker_ready && break || sleep 5; done" \
+  180
+
+# ── Step 6: Enable share groups ──────────────────────────────
+run_ssm "6/6 Enable KIP-932 share groups" \
+  "/opt/kafka/bin/kafka-features.sh --bootstrap-server localhost:9092 upgrade --feature share.version=1 2>&1" \
+  30
+
+# ── Done ─────────────────────────────────────────────────────
+BOOTSTRAP=$(aws cloudformation describe-stacks \
+  --stack-name "$BROKER_STACK" --profile "$PROFILE" --region "$REGION" \
+  --query "Stacks[0].Outputs[?OutputKey=='BootstrapServers'].OutputValue" \
+  --output text)
+
+echo ""
+echo "=== Kafka broker ready ==="
+echo "  Bootstrap servers: $BOOTSTRAP"
+echo ""
+echo "Next step: deploy the application stack"
+echo "  sam build --template stacks/3-app.yaml"
+echo "  sam deploy --stack-name kafka-queue-app \\"
+echo "    --template-file .aws-sam/build/template.yaml \\"
+echo "    --capabilities CAPABILITY_IAM --resolve-s3 \\"
+echo "    --profile $PROFILE --region $REGION"
diff --git a/smk-lambda-queue-mode-python-sam/scripts/setup-vpc-endpoints.sh b/smk-lambda-queue-mode-python-sam/scripts/setup-vpc-endpoints.sh
new file mode 100755
index 000000000..39bebaf13
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/scripts/setup-vpc-endpoints.sh
@@ -0,0 +1,108 @@
+#!/usr/bin/env bash
+# setup-vpc-endpoints.sh — Create VPC interface endpoints required for Queue mode
+#
+# ESM pollers run in private subnets and require three VPC endpoints:
+#   lambda  — to invoke the Lambda function
+#   sts     — to obtain temporary credentials
+#   sqs     — to write failed records to the OnFailure DLQ
+#
+# IMPORTANT: All three are required. Missing the SQS endpoint causes a silent
+# connection error — the ESM stays Enabled/OK but Lambda stops being invoked
+# after the first batch when a failed record triggers a DLQ write.
+#
+# This script is idempotent — it skips endpoints that already exist.
+#
+# Usage (Path A — full deploy, resolves VPC from kafka-queue-network stack):
+#   ./scripts/setup-vpc-endpoints.sh --region <region> --profile <profile>
+#
+# Usage (Path B/C — BYO VPC, provide values directly):
+#   ./scripts/setup-vpc-endpoints.sh --region <region> --profile <profile> \
+#     --vpc-id vpc-xxx \
+#     --subnet-ids "subnet-a,subnet-b,subnet-c" \
+#     --security-group-id sg-xxx
+#
+# If your VPC already has these endpoints, this script will skip them.
+
+set -euo pipefail
+
+REGION="${AWS_DEFAULT_REGION:-us-east-1}"
+PROFILE="${AWS_PROFILE:-default}"
+NETWORK_STACK="kafka-queue-network"
+VPC_ID=""
+SUBNET_IDS=""
+SG_ID=""
+
+while [[ $# -gt 0 ]]; do
+  case $1 in
+    --region)           REGION="$2";   shift 2 ;;
+    --profile)          PROFILE="$2";  shift 2 ;;
+    --vpc-id)           VPC_ID="$2";   shift 2 ;;
+    --subnet-ids)       SUBNET_IDS="$2"; shift 2 ;;
+    --security-group-id) SG_ID="$2";  shift 2 ;;
+    *) echo "Unknown option: $1"; exit 1 ;;
+  esac
+done
+
+echo "=== Setting up VPC endpoints ==="
+echo "  Region:  $REGION"
+echo "  Profile: $PROFILE"
+echo ""
+
+# ── Resolve from stack outputs if not provided ────────────────
+if [ -z "$VPC_ID" ]; then
+  echo "Resolving VPC config from stack $NETWORK_STACK..."
+  get_output() {
+    aws cloudformation describe-stacks \
+      --stack-name "$NETWORK_STACK" --profile "$PROFILE" --region "$REGION" \
+      --query "Stacks[0].Outputs[?OutputKey=='$1'].OutputValue" \
+      --output text
+  }
+  VPC_ID=$(get_output "VpcId")
+  SUBNET_A=$(get_output "PrivateSubnetA")
+  SUBNET_B=$(get_output "PrivateSubnetB")
+  SUBNET_C=$(get_output "PrivateSubnetC")
+  SG_ID=$(get_output "LambdaSecurityGroupId")
+  SUBNET_IDS="$SUBNET_A $SUBNET_B $SUBNET_C"
+else
+  echo "Using provided VPC config..."
+  # Convert comma-separated to space-separated for aws CLI
+  SUBNET_IDS="${SUBNET_IDS//,/ }"
+fi
+
+echo "  VPC:     $VPC_ID"
+echo "  Subnets: $SUBNET_IDS"
+echo "  SG:      $SG_ID"
+echo ""
+
+# ── Create endpoints ─────────────────────────────────────────
+for SVC in lambda sts sqs; do
+  EXISTING=$(aws ec2 describe-vpc-endpoints \
+    --profile "$PROFILE" --region "$REGION" \
+    --filters \
+      "Name=service-name,Values=com.amazonaws.$REGION.$SVC" \
+      "Name=vpc-id,Values=$VPC_ID" \
+      "Name=vpc-endpoint-state,Values=available,pending" \
+    --query 'VpcEndpoints[0].VpcEndpointId' \
+    --output text 2>/dev/null)
+
+  if [ "$EXISTING" = "None" ] || [ -z "$EXISTING" ]; then
+    echo "Creating $SVC endpoint..."
+    EPID=$(aws ec2 create-vpc-endpoint \
+      --vpc-id "$VPC_ID" \
+      --service-name "com.amazonaws.$REGION.$SVC" \
+      --vpc-endpoint-type Interface \
+      --subnet-ids $SUBNET_IDS \
+      --security-group-ids "$SG_ID" \
+      --private-dns-enabled \
+      --profile "$PROFILE" --region "$REGION" \
+      --query 'VpcEndpoint.VpcEndpointId' \
+      --output text)
+    echo "  Created: $EPID"
+  else
+    echo "$SVC endpoint already exists: $EXISTING (skipping)"
+  fi
+done
+
+echo ""
+echo "All VPC endpoints in place. Wait ~30s for endpoints to become available,"
+echo "then run ./scripts/create-esm.sh to create the Queue mode ESM."
diff --git a/smk-lambda-queue-mode-python-sam/src/producer/producer.py b/smk-lambda-queue-mode-python-sam/src/producer/producer.py
new file mode 100644
index 000000000..f4b587df0
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/src/producer/producer.py
@@ -0,0 +1,74 @@
+"""KQD Producer Lambda — publishes work items to Kafka topic.
+
+PLAINTEXT, no auth. Every 7th item (taskIndex % 7 == 0) has shouldFail=True
+to demonstrate the RELEASE/retry path.
+"""
+
+import json
+import logging
+import os
+import uuid
+
+from confluent_kafka import KafkaException, Producer
+from confluent_kafka.admin import AdminClient, NewTopic
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+TOPIC = os.environ.get("KAFKA_TOPIC", "kafka-queue-task-worker")
+BOOTSTRAP_SERVERS = os.environ.get("BOOTSTRAP_SERVERS", "")
+DEFAULT_COUNT = 50
+
+
+def _config():
+    return {"bootstrap.servers": BOOTSTRAP_SERVERS}
+
+
+def _ensure_topic(admin, topic):
+    nt = NewTopic(topic, num_partitions=3, replication_factor=1)
+    for t, future in admin.create_topics([nt]).items():
+        try:
+            future.result()
+            logger.info("Created topic %s", t)
+        except KafkaException as e:
+            if "already exists" in str(e).lower():
+                logger.info("Topic %s already exists", t)
+            else:
+                raise
+
+
+def lambda_handler(event, context):
+    event = event or {}
+    count = int(event.get("count", DEFAULT_COUNT))
+    topic = event.get("topic") or TOPIC
+
+    _ensure_topic(AdminClient(_config()), topic)
+
+    producer = Producer(_config())
+    failures = 0
+
+    def on_delivery(err, _):
+        nonlocal failures
+        if err:
+            failures += 1
+            logger.error("Delivery failed: %s", err)
+
+    for i in range(count):
+        item = {
+            "jobId": str(uuid.uuid4()),
+            "taskIndex": i,
+            "payload": f"task-{i}",
+            "shouldFail": i % 7 == 0,
+        }
+        producer.produce(
+            topic,
+            key=item["jobId"].encode(),
+            value=json.dumps(item).encode(),
+            on_delivery=on_delivery,
+        )
+        producer.poll(0)
+
+    producer.flush()
+    result = {"topic": topic, "produced": count, "failures": failures}
+    logger.info("Producer done: %s", json.dumps(result))
+    return result
diff --git a/smk-lambda-queue-mode-python-sam/src/producer/requirements.txt b/smk-lambda-queue-mode-python-sam/src/producer/requirements.txt
new file mode 100644
index 000000000..8f821ed41
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/src/producer/requirements.txt
@@ -0,0 +1 @@
+confluent-kafka>=2.5.0,<3.0.0
diff --git a/smk-lambda-queue-mode-python-sam/src/worker/handler.py b/smk-lambda-queue-mode-python-sam/src/worker/handler.py
new file mode 100644
index 000000000..820194979
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/src/worker/handler.py
@@ -0,0 +1,44 @@
+"""KQD Worker Lambda -- Kafka Queue mode (KIP-932) consumer."""
+
+import base64
+import json
+import logging
+import time
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+
+def lambda_handler(event, context):
+    failures = []
+
+    for tp_key, records in event.get("records", {}).items():
+        for r in records:
+            try:
+                payload = json.loads(base64.b64decode(r["value"]).decode("utf-8"))
+            except Exception:
+                payload = {"raw": r.get("value", "")}
+
+            logger.info(
+                "KAFKA_RECORD topic=%s partition=%s offset=%s payload=%s",
+                r.get("topic"),
+                r.get("partition"),
+                r.get("offset"),
+                json.dumps(payload),
+            )
+
+            if payload.get("shouldFail"):
+                identifier = f"{r['topic']}-{r['partition']}-{r['offset']}"
+                logger.warning("Simulated failure, releasing record: %s", identifier)
+                failures.append({"itemIdentifier": identifier})
+            else:
+                # 0.5s simulated processing keeps records inflight long enough
+                # to observe concurrent pollers during the scaling test
+                time.sleep(0.5)
+
+    logger.info(
+        "Batch done: %d record(s), %d failure(s)",
+        sum(len(v) for v in event.get("records", {}).values()),
+        len(failures),
+    )
+    return {"batchItemFailures": failures}
diff --git a/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml
new file mode 100644
index 000000000..ba4c12b0a
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml
@@ -0,0 +1,196 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Description: Stack 1 - Networking (VPC, subnets, IGW, security groups)
+
+Resources:
+
+  VPC:
+    Type: AWS::EC2::VPC
+    Properties:
+      CidrBlock: 10.0.0.0/16
+      EnableDnsSupport: true
+      EnableDnsHostnames: true
+      Tags:
+        - Key: Name
+          Value: kafka-queue-vpc
+
+  InternetGateway:
+    Type: AWS::EC2::InternetGateway
+    Properties:
+      Tags:
+        - Key: Name
+          Value: kafka-queue-igw
+
+  VPCGatewayAttachment:
+    Type: AWS::EC2::VPCGatewayAttachment
+    Properties:
+      VpcId: !Ref VPC
+      InternetGatewayId: !Ref InternetGateway
+
+  PublicSubnet:
+    Type: AWS::EC2::Subnet
+    Properties:
+      VpcId: !Ref VPC
+      CidrBlock: 10.0.0.0/24
+      AvailabilityZone: !Select [0, !GetAZs ""]
+      MapPublicIpOnLaunch: true
+      Tags:
+        - Key: Name
+          Value: kafka-queue-public
+
+  PublicRouteTable:
+    Type: AWS::EC2::RouteTable
+    Properties:
+      VpcId: !Ref VPC
+      Tags:
+        - Key: Name
+          Value: kafka-queue-public-rt
+
+  PublicRoute:
+    Type: AWS::EC2::Route
+    DependsOn: VPCGatewayAttachment
+    Properties:
+      RouteTableId: !Ref PublicRouteTable
+      DestinationCidrBlock: 0.0.0.0/0
+      GatewayId: !Ref InternetGateway
+
+  PublicSubnetRTAssociation:
+    Type: AWS::EC2::SubnetRouteTableAssociation
+    Properties:
+      SubnetId: !Ref PublicSubnet
+      RouteTableId: !Ref PublicRouteTable
+
+  PrivateSubnetA:
+    Type: AWS::EC2::Subnet
+    Properties:
+      VpcId: !Ref VPC
+      CidrBlock: 10.0.1.0/24
+      AvailabilityZone: !Select [0, !GetAZs ""]
+      Tags:
+        - Key: Name
+          Value: kafka-queue-private-a
+
+  PrivateSubnetB:
+    Type: AWS::EC2::Subnet
+    Properties:
+      VpcId: !Ref VPC
+      CidrBlock: 10.0.2.0/24
+      AvailabilityZone: !Select [1, !GetAZs ""]
+      Tags:
+        - Key: Name
+          Value: kafka-queue-private-b
+
+  PrivateSubnetC:
+    Type: AWS::EC2::Subnet
+    Properties:
+      VpcId: !Ref VPC
+      CidrBlock: 10.0.3.0/24
+      AvailabilityZone: !Select [2, !GetAZs ""]
+      Tags:
+        - Key: Name
+          Value: kafka-queue-private-c
+
+  BrokerSecurityGroup:
+    Type: AWS::EC2::SecurityGroup
+    Properties:
+      GroupDescription: Kafka broker - PLAINTEXT 9092 from VPC
+      VpcId: !Ref VPC
+      SecurityGroupIngress:
+        - IpProtocol: tcp
+          FromPort: 9092
+          ToPort: 9092
+          CidrIp: 10.0.0.0/16
+          Description: Kafka PLAINTEXT from VPC
+      Tags:
+        - Key: Name
+          Value: kafka-queue-broker-sg
+
+  LambdaSecurityGroup:
+    Type: AWS::EC2::SecurityGroup
+    Properties:
+      GroupDescription: Lambda functions - egress to Kafka 9092 and HTTPS
+      VpcId: !Ref VPC
+      SecurityGroupEgress:
+        - IpProtocol: tcp
+          FromPort: 9092
+          ToPort: 9092
+          CidrIp: 10.0.0.0/16
+          Description: Kafka PLAINTEXT
+        - IpProtocol: tcp
+          FromPort: 443
+          ToPort: 443
+          CidrIp: 0.0.0.0/0
+          Description: HTTPS for AWS APIs and VPC endpoints
+      Tags:
+        - Key: Name
+          Value: kafka-queue-lambda-sg
+
+  # Allow inbound 443 from VPC (VPC endpoint ENIs need this)
+  LambdaSGIngressVPC:
+    Type: AWS::EC2::SecurityGroupIngress
+    Properties:
+      GroupId: !Ref LambdaSecurityGroup
+      IpProtocol: tcp
+      FromPort: 443
+      ToPort: 443
+      CidrIp: 10.0.0.0/16
+      Description: VPC endpoints HTTPS inbound
+
+  # Self-referencing rule so Hyperplane ENIs can reach endpoint ENIs
+  LambdaSGIngressSelf:
+    Type: AWS::EC2::SecurityGroupIngress
+    Properties:
+      GroupId: !Ref LambdaSecurityGroup
+      IpProtocol: tcp
+      FromPort: 443
+      ToPort: 443
+      SourceSecurityGroupId: !Ref LambdaSecurityGroup
+      Description: Self - Hyperplane ENI to VPC endpoint ENI
+
+  # VPC Interface Endpoints — all three are required for Queue mode ESM pollers:
+  #   lambda  — pollers invoke the Lambda function
+  #   sts     — pollers obtain temporary credentials
+  #   sqs     — pollers write failed records to the OnFailure DLQ
+  #
+  # IMPORTANT: Missing the SQS endpoint causes a silent connection error —
+  # the ESM stays Enabled/OK but Lambda stops being invoked after the first
+  # batch when a failed record triggers a DLQ write.
+  #
+  # These are created via scripts/setup-vpc-endpoints.sh rather than as
+  # CloudFormation resources because AWS::EC2::VpcEndpoint is restricted
+  # in some accounts (e.g. Isengard/bindled accounts used for testing).
+
+Outputs:
+  VpcId:
+    Value: !Ref VPC
+    Export:
+      Name: kafka-queue-VpcId
+
+  PublicSubnetId:
+    Value: !Ref PublicSubnet
+    Export:
+      Name: kafka-queue-PublicSubnetId
+
+  PrivateSubnetA:
+    Value: !Ref PrivateSubnetA
+    Export:
+      Name: kafka-queue-PrivateSubnetA
+
+  PrivateSubnetB:
+    Value: !Ref PrivateSubnetB
+    Export:
+      Name: kafka-queue-PrivateSubnetB
+
+  PrivateSubnetC:
+    Value: !Ref PrivateSubnetC
+    Export:
+      Name: kafka-queue-PrivateSubnetC
+
+  BrokerSecurityGroupId:
+    Value: !Ref BrokerSecurityGroup
+    Export:
+      Name: kafka-queue-BrokerSecurityGroupId
+
+  LambdaSecurityGroupId:
+    Value: !Ref LambdaSecurityGroup
+    Export:
+      Name: kafka-queue-LambdaSecurityGroupId
diff --git a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml
new file mode 100644
index 000000000..01d42a003
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml
@@ -0,0 +1,55 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Description: Stack 2 - Self-managed Kafka 4.2.x broker on EC2 (KRaft, PLAINTEXT)
+
+Parameters:
+  InstanceType:
+    Type: String
+    Default: t3.medium
+  LatestAL2023AmiId:
+    Type: AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>
+    Default: /aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-x86_64
+
+Resources:
+
+  KafkaInstanceRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: '2012-10-17'
+        Statement:
+          - Effect: Allow
+            Principal:
+              Service: ec2.amazonaws.com
+            Action: sts:AssumeRole
+      ManagedPolicyArns:
+        - arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore
+
+  KafkaInstanceProfile:
+    Type: AWS::IAM::InstanceProfile
+    Properties:
+      Roles:
+        - !Ref KafkaInstanceRole
+
+  KafkaInstance:
+    Type: AWS::EC2::Instance
+    Properties:
+      InstanceType: !Ref InstanceType
+      ImageId: !Ref LatestAL2023AmiId
+      SubnetId: !ImportValue kafka-queue-PublicSubnetId
+      SecurityGroupIds:
+        - !ImportValue kafka-queue-BrokerSecurityGroupId
+      IamInstanceProfile: !Ref KafkaInstanceProfile
+      Tags:
+        - Key: Name
+          Value: kafka-queue-kafka-broker
+
+Outputs:
+  BootstrapServers:
+    Value: !Sub "${KafkaInstance.PrivateIp}:9092"
+    Export:
+      Name: kafka-queue-BootstrapServers
+
+  BrokerInstanceId:
+    Value: !Ref KafkaInstance
+    Export:
+      Name: kafka-queue-BrokerInstanceId
diff --git a/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml b/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml
new file mode 100644
index 000000000..a2e07d0f2
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml
@@ -0,0 +1,143 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+Description: >-
+  Stack 3 - Worker Lambda, Producer Lambda, and SQS DLQ
+  for the Kafka Queue mode (KIP-932) pattern.
+  Supports three deployment paths:
+    - Full: imports VPC outputs from kafka-queue-network stack (set UseExistingInfra=false)
+    - Bring-your-own Kafka: provide BootstrapServers, VpcSubnetIds, VpcSecurityGroupId
+    - Bring-your-own Kafka + VPC: same as above (UseExistingInfra=true)
+
+Parameters:
+  KafkaTopic:
+    Type: String
+    Default: kafka-queue-task-worker
+    Description: Kafka topic name the worker consumes and the producer publishes to.
+
+  BootstrapServers:
+    Type: String
+    Default: ""
+    Description: >-
+      Kafka bootstrap servers (e.g. 10.0.0.43:9092).
+      Leave blank to import from kafka-queue-broker stack output.
+
+  VpcSubnetIds:
+    Type: String
+    Default: ""
+    Description: >-
+      Comma-separated private subnet IDs for the producer Lambda VPC config.
+      Leave blank to import from kafka-queue-network stack output.
+
+  VpcSecurityGroupId:
+    Type: String
+    Default: ""
+    Description: >-
+      Security group ID for the producer Lambda.
+      Leave blank to import from kafka-queue-network stack output.
+
+  UseExistingInfra:
+    Type: String
+    Default: "false"
+    AllowedValues: ["true", "false"]
+    Description: >-
+      Set to true when providing your own BootstrapServers, VpcSubnetIds,
+      and VpcSecurityGroupId (skips CloudFormation imports from kafka-queue-network
+      and kafka-queue-broker stacks).
+
+Conditions:
+  ImportInfra: !Equals [!Ref UseExistingInfra, "false"]
+
+Resources:
+
+  # SQS DLQ — receives records that exhaust their delivery attempt limit.
+  # group.share.delivery.count.limit on the broker and MaximumRetryAttempts
+  # on the ESM must be set to the same value.
+  TaskWorkerDLQ:
+    Type: AWS::SQS::Queue
+    Properties:
+      QueueName: !Sub ${AWS::StackName}-dlq
+      MessageRetentionPeriod: 1209600
+      VisibilityTimeout: 300
+      SqsManagedSseEnabled: true
+
+  WorkerFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      FunctionName: !Sub ${AWS::StackName}-worker
+      Runtime: python3.12
+      CodeUri: ../src/worker/
+      Handler: handler.lambda_handler
+      MemorySize: 512
+      Timeout: 25
+      Policies:
+        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
+        - Version: '2012-10-17'
+          Statement:
+            - Effect: Allow
+              Action: sqs:SendMessage
+              Resource: !GetAtt TaskWorkerDLQ.Arn
+            - Effect: Allow
+              Action:
+                - ec2:CreateNetworkInterface
+                - ec2:DescribeNetworkInterfaces
+                - ec2:DeleteNetworkInterface
+                - ec2:DescribeVpcs
+                - ec2:DescribeSubnets
+                - ec2:DescribeSecurityGroups
+              Resource: '*'
+
+  ProducerFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      FunctionName: !Sub ${AWS::StackName}-producer
+      Runtime: python3.12
+      CodeUri: ../src/producer/
+      Handler: producer.lambda_handler
+      MemorySize: 256
+      Timeout: 60
+      VpcConfig:
+        SubnetIds: !Split
+          - ","
+          - !If
+            - ImportInfra
+            - !Join
+              - ","
+              - - !ImportValue kafka-queue-PrivateSubnetA
+                - !ImportValue kafka-queue-PrivateSubnetB
+                - !ImportValue kafka-queue-PrivateSubnetC
+            - !Ref VpcSubnetIds
+        SecurityGroupIds:
+          - !If
+            - ImportInfra
+            - !ImportValue kafka-queue-LambdaSecurityGroupId
+            - !Ref VpcSecurityGroupId
+      Environment:
+        Variables:
+          KAFKA_TOPIC: !Ref KafkaTopic
+          BOOTSTRAP_SERVERS: !If
+            - ImportInfra
+            - !ImportValue kafka-queue-BootstrapServers
+            - !Ref BootstrapServers
+      Policies:
+        - arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole
+
+Outputs:
+  WorkerFunctionArn:
+    Value: !GetAtt WorkerFunction.Arn
+    Export:
+      Name: kafka-queue-WorkerFunctionArn
+
+  WorkerFunctionName:
+    Value: !Ref WorkerFunction
+    Export:
+      Name: kafka-queue-WorkerFunctionName
+
+  TaskWorkerDLQArn:
+    Value: !GetAtt TaskWorkerDLQ.Arn
+    Export:
+      Name: kafka-queue-TaskWorkerDLQArn
+
+  TaskWorkerDLQUrl:
+    Value: !Ref TaskWorkerDLQ
+    Export:
+      Name: kafka-queue-TaskWorkerDLQUrl
diff --git a/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml b/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml
new file mode 100644
index 000000000..41dd53d6a
--- /dev/null
+++ b/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml
@@ -0,0 +1,74 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Description: Stack 4 - CloudWatch alarms and dashboard for Queue mode ESM
+
+Parameters:
+  ESMUuid:
+    Type: String
+    Description: UUID of the Queue mode Event Source Mapping (from create-esm.sh output)
+
+Resources:
+
+  DlqDeliveryAlarm:
+    Type: AWS::CloudWatch::Alarm
+    Properties:
+      AlarmName: kafka-queue-dlq-delivery
+      AlarmDescription: Records routed to DLQ (delivery attempts exhausted)
+      Namespace: AWS/Lambda
+      MetricName: OnFailureDestinationDeliveredEventCount
+      Dimensions:
+        - Name: EventSourceMappingUUID
+          Value: !Ref ESMUuid
+      Statistic: Sum
+      Period: 60
+      EvaluationPeriods: 1
+      Threshold: 1
+      ComparisonOperator: GreaterThanOrEqualToThreshold
+      TreatMissingData: notBreaching
+
+  LagGrowthAlarm:
+    Type: AWS::CloudWatch::Alarm
+    Properties:
+      AlarmName: kafka-queue-lag-growth
+      AlarmDescription: Share group lag is growing
+      Namespace: AWS/Lambda
+      MetricName: MaxShareGroupLag
+      Dimensions:
+        - Name: EventSourceMappingUUID
+          Value: !Ref ESMUuid
+      Statistic: Maximum
+      Period: 300
+      EvaluationPeriods: 2
+      Threshold: 1000
+      ComparisonOperator: GreaterThanThreshold
+      TreatMissingData: notBreaching
+
+  PollerErrorAlarm:
+    Type: AWS::CloudWatch::Alarm
+    Properties:
+      AlarmName: kafka-queue-poller-errors
+      AlarmDescription: ESM reported polling errors
+      Namespace: AWS/Lambda
+      MetricName: PollingErrorCount
+      Dimensions:
+        - Name: EventSourceMappingUUID
+          Value: !Ref ESMUuid
+      Statistic: Sum
+      Period: 60
+      EvaluationPeriods: 1
+      Threshold: 1
+      ComparisonOperator: GreaterThanOrEqualToThreshold
+      TreatMissingData: notBreaching
+
+  KafkaQueueDashboard:
+    Type: AWS::CloudWatch::Dashboard
+    Properties:
+      DashboardName: kafka-queue-dashboard
+      DashboardBody: !Sub
+        - |
+          {"widgets":[
+            {"type":"metric","x":0,"y":0,"width":12,"height":6,"properties":{"title":"Share Group Lag","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","SumShareGroupLag","EventSourceMappingUUID","${ESM}"],["AWS/Lambda","MaxShareGroupLag","EventSourceMappingUUID","${ESM}"]]}},
+            {"type":"metric","x":12,"y":0,"width":12,"height":6,"properties":{"title":"Provisioned Pollers","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","ProvisionedPollers","EventSourceMappingUUID","${ESM}"]]}},
+            {"type":"metric","x":0,"y":6,"width":12,"height":6,"properties":{"title":"Event Counts","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","PolledEventCount","EventSourceMappingUUID","${ESM}"],["AWS/Lambda","InvokedEventCount","EventSourceMappingUUID","${ESM}"],["AWS/Lambda","OnFailureDestinationDeliveredEventCount","EventSourceMappingUUID","${ESM}"]]}},
+            {"type":"metric","x":12,"y":6,"width":12,"height":6,"properties":{"title":"Errors","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","PollingErrorCount","EventSourceMappingUUID","${ESM}"],["AWS/Lambda","FailedInvokeEventCount","EventSourceMappingUUID","${ESM}"]]}}
+          ]}
+        - ESM: !Ref ESMUuid