From 40c4cae474c952421fa280b51c4a52696ce3b249 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Wed, 16 Sep 2026 12:48:36 -0400 Subject: [PATCH 01/28] feat: add smk-lambda-queue-mode-python-sam pattern Serverless Land pattern for Kafka Queue mode (KIP-932) with self-managed Apache Kafka 4.2+ and AWS Lambda. Pattern includes: - 4 CloudFormation stacks (network, broker, app, observability) - Python 3.12 worker Lambda with partial batch response - Python 3.12 producer Lambda for testing - create-esm.sh script (ConsumptionMode not yet in SAM/CFN schema) - 3 deployment paths: full / bring-your-own-Kafka / bring-your-own-Kafka+VPC - CloudWatch dashboard and alarms - Local test event Author: Vaibhav Jain (AWS Senior Delivery Consultant) --- smk-lambda-queue-mode-python-sam/.gitignore | 5 + smk-lambda-queue-mode-python-sam/README.md | 275 ++++++++++++++++++ .../events/kafka-event.json | 40 +++ .../example-pattern.json | 69 +++++ .../scripts/create-esm.sh | 139 +++++++++ .../src/producer/producer.py | 72 +++++ .../src/producer/requirements.txt | 1 + .../src/worker/handler.py | 44 +++ .../stacks/1-network.yaml | 192 ++++++++++++ .../stacks/2-broker.yaml | 162 +++++++++++ .../stacks/3-app.yaml | 165 +++++++++++ .../stacks/4-observability.yaml | 86 ++++++ 12 files changed, 1250 insertions(+) create mode 100644 smk-lambda-queue-mode-python-sam/.gitignore create mode 100644 smk-lambda-queue-mode-python-sam/README.md create mode 100644 smk-lambda-queue-mode-python-sam/events/kafka-event.json create mode 100644 smk-lambda-queue-mode-python-sam/example-pattern.json create mode 100755 smk-lambda-queue-mode-python-sam/scripts/create-esm.sh create mode 100644 smk-lambda-queue-mode-python-sam/src/producer/producer.py create mode 100644 smk-lambda-queue-mode-python-sam/src/producer/requirements.txt create mode 100644 smk-lambda-queue-mode-python-sam/src/worker/handler.py create mode 100644 smk-lambda-queue-mode-python-sam/stacks/1-network.yaml create mode 100644 smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml create mode 100644 smk-lambda-queue-mode-python-sam/stacks/3-app.yaml create mode 100644 smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml 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..1f89465eb --- /dev/null +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -0,0 +1,275 @@ +# 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 + +``` +Producer Lambda ──► Kafka topic (3 partitions) + │ + ┌──────────┼──────────┐ + Poller 1 Poller 2 Poller 3..10 + │ │ │ + (same partition can be served by multiple pollers) + └──────────┼──────────┘ + │ + Worker Lambda + ├── DynamoDB (idempotency) + └── SQS DLQ (failed records) +``` + +## 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, DynamoDB, 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 kqd-network \ + --template-file stacks/1-network.yaml \ + --region +``` + +**Step 2: Deploy the Kafka broker** + +```bash +aws cloudformation deploy \ + --stack-name kqd-broker \ + --template-file stacks/2-broker.yaml \ + --capabilities CAPABILITY_IAM \ + --region +``` + +This provisions a t3.medium EC2 instance running Apache Kafka 4.2.x in KRaft mode with share groups enabled. The stack signals CloudFormation when Kafka is ready (~10 minutes). + +**Step 3: Build and deploy the application** + +```bash +sam build --template stacks/3-app.yaml +sam deploy \ + --stack-name kqd-app \ + --template-file .aws-sam/build/template.yaml \ + --capabilities CAPABILITY_IAM \ + --resolve-s3 \ + --region +``` + +**Step 4: Deploy observability (optional)** + +```bash +aws cloudformation deploy \ + --stack-name kqd-observability \ + --template-file stacks/4-observability.yaml \ + --region +``` + +--- + +### Path B: Bring your own Kafka cluster + +Skip Steps 1 and 2. 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 kqd-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 +``` + +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 + +--- + +### Path C: Bring your own Kafka cluster and VPC + +Same as Path B — `UseExistingInfra=true` handles both cases. + +--- + +## Create the Queue mode ESM + +After deploying the application stack, create the Event Source Mapping with `ConsumptionMode: Queue`: + +```bash +chmod +x scripts/create-esm.sh +./scripts/create-esm.sh --region --profile +``` + +Wait ~60 seconds for the ESM to reach `State: Enabled`. + +> **Note:** The script uses `curl --aws-sigv4` to call the Lambda API directly because `ConsumptionMode: Queue` is not yet in the SAM or AWS CLI service model. Update to the latest AWS CLI or SAM when this field becomes available to use standard tooling. + +--- + +## Testing + +**Produce 20 records:** + +```bash +aws lambda invoke \ + --function-name kqd-app-producer \ + --region \ + --cli-binary-format raw-in-base64-out \ + --payload '{"count": 20}' /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/kqd-app-worker \ + --follow \ + --filter-pattern KAFKA_RECORD \ + --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. + +**Test locally:** + +```bash +sam local invoke WorkerFunction \ + --template stacks/3-app.yaml \ + --event events/kafka-event.json +``` + +--- + +## Verifying Queue mode behavior + +**Confirm the share group exists on the broker:** + +```bash +# On the Kafka broker (via SSM or SSH) +bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --list +# Should show your consumer group ID + +bin/kafka-share-groups.sh --bootstrap-server localhost:9092 \ + --describe --group kqd-queue-group- +# Shows per-partition lag with multiple pollers assigned +``` + +**Confirm share groups are enabled:** + +```bash +bin/kafka-features.sh --bootstrap-server localhost:9092 describe | grep share +# Should show: FinalizedVersionLevel: 1 +``` + +--- + +## 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 --region + +# 2. Delete application stacks +aws cloudformation delete-stack --stack-name kqd-observability --region +aws cloudformation delete-stack --stack-name kqd-app --region + +# 3. Delete broker (if deployed) +aws cloudformation delete-stack --stack-name kqd-broker --region + +# 4. Delete network (if deployed) +aws cloudformation delete-stack --stack-name kqd-network --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 | + +## Author + +**Vaibhav Jain** +AWS — Senior Delivery Consultant +[LinkedIn](https://www.linkedin.com/in/vaibhavjainv/) 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..b3f0d61fe --- /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": { + "kqd-task-worker-0": [ + { + "topic": "kqd-task-worker", + "partition": 0, + "offset": 0, + "timestamp": 1726000000000, + "timestampType": "CREATE_TIME", + "key": "am9iSWQtMQ==", + "value": "eyJqb2JJZCI6ICJqb2JJZC0xIiwgInRhc2tJbmRleCI6IDEsICJwYXlsb2FkIjogInRhc2stMSIsICJzaG91bGRGYWlsIjogZmFsc2V9", + "headers": [] + }, + { + "topic": "kqd-task-worker", + "partition": 0, + "offset": 1, + "timestamp": 1726000001000, + "timestampType": "CREATE_TIME", + "key": "am9iSWQtMg==", + "value": "eyJqb2JJZCI6ICJqb2JJZC0yIiwgInRhc2tJbmRleCI6IDcsICJwYXlsb2FkIjogInRhc2stNyIsICJzaG91bGRGYWlsIjogdHJ1ZX0=", + "headers": [] + } + ], + "kqd-task-worker-1": [ + { + "topic": "kqd-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..0f318dcb7 --- /dev/null +++ b/smk-lambda-queue-mode-python-sam/example-pattern.json @@ -0,0 +1,69 @@ +{ + "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 KIP-932: Queues for Kafka", + "link": "https://cwiki.apache.org/confluence/display/KAFKA/KIP-932+Queues+for+Kafka" + }, + { + "text": "Lambda ESM provisioned mode for Kafka", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/kafka-scaling-modes.html" + }, + { + "text": "Partial batch response for Lambda", + "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 stacks in reverse order: aws cloudformation delete-stack --stack-name kqd-esm, then kqd-app, kqd-broker, kqd-network." + ] + }, + "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/" + } + ] +} 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..ca941fe66 --- /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 kqd-network, kqd-broker, kqd-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="kqd-app" +BROKER_STACK="kqd-broker" +NETWORK_STACK="kqd-network" +CONSUMER_GROUP_ID="kqd-queue-group-$(date +%s)" +TOPIC="kqd-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 kqd-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/kqd-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/src/producer/producer.py b/smk-lambda-queue-mode-python-sam/src/producer/producer.py new file mode 100644 index 000000000..da24a1dda --- /dev/null +++ b/smk-lambda-queue-mode-python-sam/src/producer/producer.py @@ -0,0 +1,72 @@ +"""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", "kqd-task-worker") +BOOTSTRAP_SERVERS = os.environ.get("BOOTSTRAP_SERVERS", "") +DEFAULT_COUNT = 50 + + +def _config(): + return {"bootstrap.servers": BOOTSTRAP_SERVERS} + + +def _ensure_topic(admin): + nt = NewTopic(TOPIC, num_partitions=3, replication_factor=1) + for topic, future in admin.create_topics([nt]).items(): + try: + future.result() + logger.info("Created topic %s", topic) + except KafkaException as e: + if "already exists" in str(e).lower(): + logger.info("Topic %s already exists", topic) + else: + raise + + +def lambda_handler(event, context): + event = event or {} + count = int(event.get("count", DEFAULT_COUNT)) + + _ensure_topic(AdminClient(_config())) + + 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..2d1f893a9 --- /dev/null +++ b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml @@ -0,0 +1,192 @@ +AWSTemplateFormatVersion: '2010-09-09' +Description: KQD 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: kqd-vpc + + InternetGateway: + Type: AWS::EC2::InternetGateway + Properties: + Tags: + - Key: Name + Value: kqd-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: kqd-public + + PublicRouteTable: + Type: AWS::EC2::RouteTable + Properties: + VpcId: !Ref VPC + Tags: + - Key: Name + Value: kqd-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: kqd-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: kqd-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: kqd-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: kqd-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: kqd-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 Endpoint for Lambda (allows ESM pollers to invoke Lambda) + # NOTE: Created separately via CLI if this stack is deployed fresh: + # aws ec2 create-vpc-endpoint --vpc-endpoint-type Interface \ + # --vpc-id --service-name com.amazonaws..lambda \ + # --subnet-ids --security-group-ids \ + # --private-dns-enabled --region + # Same for com.amazonaws..sts + # These endpoints are not managed here to avoid conflict on update. + +Outputs: + VpcId: + Value: !Ref VPC + Export: + Name: kqd-VpcId + + PublicSubnetId: + Value: !Ref PublicSubnet + Export: + Name: kqd-PublicSubnetId + + PrivateSubnetA: + Value: !Ref PrivateSubnetA + Export: + Name: kqd-PrivateSubnetA + + PrivateSubnetB: + Value: !Ref PrivateSubnetB + Export: + Name: kqd-PrivateSubnetB + + PrivateSubnetC: + Value: !Ref PrivateSubnetC + Export: + Name: kqd-PrivateSubnetC + + BrokerSecurityGroupId: + Value: !Ref BrokerSecurityGroup + Export: + Name: kqd-BrokerSecurityGroupId + + LambdaSecurityGroupId: + Value: !Ref LambdaSecurityGroup + Export: + Name: kqd-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..becb9b8ef --- /dev/null +++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml @@ -0,0 +1,162 @@ +AWSTemplateFormatVersion: '2010-09-09' +Description: KQD Stack 2 - Self-managed Kafka 4.2.x broker on EC2 (KRaft, PLAINTEXT) + +Parameters: + KafkaVersion: + Type: String + Default: 4.2.0 + InstanceType: + Type: String + Default: t3.medium + LatestAL2023AmiId: + Type: AWS::SSM::Parameter::Value + 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 + - arn:aws:iam::aws:policy/AmazonSSMPatchAssociation + + # Automated patching: runs AWS-RunPatchBaseline every Sunday at 2am UTC + PatchAssociation: + Type: AWS::SSM::Association + Properties: + Name: AWS-RunPatchBaseline + Targets: + - Key: tag:aws:cloudformation:stack-name + Values: + - !Ref AWS::StackName + ScheduleExpression: cron(0 2 ? * SUN *) + Parameters: + Operation: + - Install + + KafkaInstanceProfile: + Type: AWS::IAM::InstanceProfile + Properties: + Roles: + - !Ref KafkaInstanceRole + + KafkaInstance: + Type: AWS::EC2::Instance + DependsOn: VPCGatewayAttachmentDependency + Properties: + InstanceType: !Ref InstanceType + ImageId: !Ref LatestAL2023AmiId + SubnetId: !ImportValue kqd-PublicSubnetId + SecurityGroupIds: + - !ImportValue kqd-BrokerSecurityGroupId + IamInstanceProfile: !Ref KafkaInstanceProfile + Tags: + - Key: Name + Value: kqd-kafka-broker + UserData: + Fn::Base64: !Sub | + #!/bin/bash + set -euo pipefail + exec > >(tee /var/log/kafka-setup.log) 2>&1 + echo "=== Kafka ${KafkaVersion} setup $(date) ===" + + signal_cfn() { + /opt/aws/bin/cfn-signal -e $1 \ + --stack ${AWS::StackName} \ + --resource KafkaInstance \ + --region ${AWS::Region} || true + } + trap 'echo "ERROR line $LINENO"; signal_cfn 1' ERR + + dnf install -y java-21-amazon-corretto-headless + echo "Java installed." + + KAFKA_DIR=/opt/kafka + mkdir -p $KAFKA_DIR + wget -q --timeout=180 --tries=3 \ + "https://archive.apache.org/dist/kafka/${KafkaVersion}/kafka_2.13-${KafkaVersion}.tgz" \ + -O /tmp/kafka.tgz + tar -xzf /tmp/kafka.tgz -C $KAFKA_DIR --strip-components=1 + echo "Kafka extracted." + + INSTANCE_IP=$(curl -s http://169.254.169.254/latest/meta-data/local-ipv4) + mkdir -p /var/kafka-logs + + PROPS=/tmp/kraft-server.properties + 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" \ + > $PROPS + + CLUSTER_ID=$($KAFKA_DIR/bin/kafka-storage.sh random-uuid) + $KAFKA_DIR/bin/kafka-storage.sh format -t $CLUSTER_ID -c $PROPS + echo "KRaft storage formatted." + + export KAFKA_HEAP_OPTS="-Xmx512m -Xms256m" + nohup $KAFKA_DIR/bin/kafka-server-start.sh $PROPS \ + > /var/log/kafka.log 2>&1 & + echo "Kafka started (PID $!)" + + echo "Waiting for broker..." + for i in $(seq 1 24); do + $KAFKA_DIR/bin/kafka-broker-api-versions.sh \ + --bootstrap-server localhost:9092 > /dev/null 2>&1 \ + && echo "Broker ready after ${!i}x5s" && break \ + || sleep 5 + done + + # Enable KIP-932 share groups + $KAFKA_DIR/bin/kafka-features.sh \ + --bootstrap-server localhost:9092 \ + upgrade --feature share.version=1 + echo "Share groups (KIP-932) enabled." + + echo "=== Setup complete $(date). Bootstrap: $INSTANCE_IP:9092 ===" + signal_cfn 0 + + CreationPolicy: + ResourceSignal: + Timeout: PT20M + + # Dummy resource to ensure IGW is attached before EC2 launches + VPCGatewayAttachmentDependency: + Type: AWS::CloudFormation::WaitConditionHandle + +Outputs: + BootstrapServers: + Value: !Sub "${KafkaInstance.PrivateIp}:9092" + Export: + Name: kqd-BootstrapServers + + BrokerInstanceId: + Value: !Ref KafkaInstance + Export: + Name: kqd-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..69534b1db --- /dev/null +++ b/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml @@ -0,0 +1,165 @@ +AWSTemplateFormatVersion: '2010-09-09' +Transform: AWS::Serverless-2016-10-31 +Description: >- + KQD Stack 3 - Worker Lambda, Producer Lambda, DynamoDB idempotency table, + and SQS DLQ for the Kafka Queue mode (KIP-932) pattern. + Supports three deployment paths: + - Full: imports VPC outputs from kqd-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: kqd-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 kqd-broker stack output. + + VpcSubnetIds: + Type: String + Default: "" + Description: >- + Comma-separated private subnet IDs for the producer Lambda VPC config. + Leave blank to import from kqd-network stack output. + + VpcSecurityGroupId: + Type: String + Default: "" + Description: >- + Security group ID for the producer Lambda. + Leave blank to import from kqd-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 kqd-network + and kqd-broker stacks). + +Conditions: + ImportInfra: !Equals [!Ref UseExistingInfra, "false"] + UseProvidedInfra: !Equals [!Ref UseExistingInfra, "true"] + +Resources: + + IdempotencyTable: + Type: AWS::DynamoDB::Table + Properties: + TableName: !Sub ${AWS::StackName}-idempotency + BillingMode: PAY_PER_REQUEST + AttributeDefinitions: + - AttributeName: id + AttributeType: S + KeySchema: + - AttributeName: id + KeyType: HASH + TimeToLiveSpecification: + AttributeName: expiration + Enabled: true + + 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 + Environment: + Variables: + IDEMPOTENCY_TABLE: !Ref IdempotencyTable + Policies: + - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole + - Version: '2012-10-17' + Statement: + - Effect: Allow + Action: + - dynamodb:PutItem + - dynamodb:GetItem + - dynamodb:DeleteItem + Resource: !GetAtt IdempotencyTable.Arn + - 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 kqd-PrivateSubnetA + - !ImportValue kqd-PrivateSubnetB + - !ImportValue kqd-PrivateSubnetC + - !Ref VpcSubnetIds + SecurityGroupIds: + - !If + - ImportInfra + - !ImportValue kqd-LambdaSecurityGroupId + - !Ref VpcSecurityGroupId + Environment: + Variables: + KAFKA_TOPIC: !Ref KafkaTopic + BOOTSTRAP_SERVERS: !If + - ImportInfra + - !ImportValue kqd-BootstrapServers + - !Ref BootstrapServers + Policies: + - arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole + +Outputs: + WorkerFunctionArn: + Value: !GetAtt WorkerFunction.Arn + Export: + Name: kqd-WorkerFunctionArn + + WorkerFunctionName: + Value: !Ref WorkerFunction + Export: + Name: kqd-WorkerFunctionName + + TaskWorkerDLQArn: + Value: !GetAtt TaskWorkerDLQ.Arn + Export: + Name: kqd-TaskWorkerDLQArn + + TaskWorkerDLQUrl: + Value: !Ref TaskWorkerDLQ + Export: + Name: kqd-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..2dfec3ffa --- /dev/null +++ b/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml @@ -0,0 +1,86 @@ +AWSTemplateFormatVersion: '2010-09-09' +Description: KQD Stack 4 - CloudWatch alarms and dashboard for Queue mode ESM + +Resources: + + AlarmTopic: + Type: AWS::SNS::Topic + Properties: + TopicName: kqd-alarms + + DlqDeliveryAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: kqd-dlq-delivery + AlarmDescription: Records routed to DLQ (delivery attempts exhausted) + Namespace: AWS/Lambda + MetricName: OnFailureDestinationDeliveredEventCount + Dimensions: + - Name: FunctionName + Value: !ImportValue kqd-WorkerFunctionName + Statistic: Sum + Period: 60 + EvaluationPeriods: 1 + Threshold: 1 + ComparisonOperator: GreaterThanOrEqualToThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + + LagGrowthAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: kqd-lag-growth + AlarmDescription: Share group offset lag is growing + Namespace: AWS/Lambda + MetricName: MaxOffsetLag + Dimensions: + - Name: FunctionName + Value: !ImportValue kqd-WorkerFunctionName + Statistic: Maximum + Period: 300 + EvaluationPeriods: 2 + Threshold: 1000 + ComparisonOperator: GreaterThanThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + + PollerErrorAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: kqd-poller-errors + AlarmDescription: ESM reported polling errors + Namespace: AWS/Lambda + MetricName: PollingErrorCount + Dimensions: + - Name: FunctionName + Value: !ImportValue kqd-WorkerFunctionName + Statistic: Sum + Period: 60 + EvaluationPeriods: 1 + Threshold: 1 + ComparisonOperator: GreaterThanOrEqualToThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + + KafkaQueueDashboard: + Type: AWS::CloudWatch::Dashboard + Properties: + DashboardName: kqd-KafkaQueue + DashboardBody: !Sub + - | + {"widgets":[ + {"type":"metric","x":0,"y":0,"width":12,"height":6,"properties":{"title":"Offset Lag","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","SumOffsetLag","FunctionName","${Fn}"],["AWS/Lambda","MaxOffsetLag","FunctionName","${Fn}"]]}}, + {"type":"metric","x":12,"y":0,"width":12,"height":6,"properties":{"title":"Pollers","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","ProvisionedPollers","FunctionName","${Fn}"]]}}, + {"type":"metric","x":0,"y":6,"width":12,"height":6,"properties":{"title":"Event Counts","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","PolledEventCount","FunctionName","${Fn}"],["AWS/Lambda","InvokedEventCount","FunctionName","${Fn}"],["AWS/Lambda","AcknowledgedEventCount","FunctionName","${Fn}"],["AWS/Lambda","OnFailureDestinationDeliveredEventCount","FunctionName","${Fn}"]]}}, + {"type":"metric","x":12,"y":6,"width":12,"height":6,"properties":{"title":"Errors","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","PollingErrorCount","FunctionName","${Fn}"],["AWS/Lambda","InvokeErrorCount","FunctionName","${Fn}"]]}} + ]} + - Fn: !ImportValue kqd-WorkerFunctionName + +Outputs: + AlarmTopicArn: + Value: !Ref AlarmTopic + Export: + Name: kqd-AlarmTopicArn From 76c40dd7b006134a8e31ce75892ec3240bf884b2 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Wed, 16 Sep 2026 15:15:23 -0400 Subject: [PATCH 02/28] docs: add Queue mode scaling verification section to README --- smk-lambda-queue-mode-python-sam/README.md | 70 +++++++++++++++------- 1 file changed, 47 insertions(+), 23 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index 1f89465eb..e6d3b5b32 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -8,19 +8,16 @@ Learn more about this pattern at Serverless Land: https://serverlessland.com/pat ## Architecture +```mermaid +graph LR + Producer["Producer\nLambda"] -->|publish| Kafka["Apache Kafka\n4.2+ Cluster"] + Kafka -->|Queue mode ESM\nConsumptionMode: Queue| Worker["Worker\nLambda"] + Worker --> DDB["DynamoDB\nIdempotency"] + Worker --> SQS["SQS\nDLQ"] + Worker --> CW["CloudWatch\nMetrics"] ``` -Producer Lambda ──► Kafka topic (3 partitions) - │ - ┌──────────┼──────────┐ - Poller 1 Poller 2 Poller 3..10 - │ │ │ - (same partition can be served by multiple pollers) - └──────────┼──────────┘ - │ - Worker Lambda - ├── DynamoDB (idempotency) - └── SQS DLQ (failed records) -``` + +**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 @@ -184,27 +181,54 @@ sam local invoke WorkerFunction \ --- -## Verifying Queue mode behavior +## 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: -**Confirm the share group exists on the broker:** +**Step 1: Produce a large batch** ```bash -# On the Kafka broker (via SSM or SSH) -bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --list -# Should show your consumer group ID +aws lambda invoke \ + --function-name kqd-app-producer \ + --region \ + --cli-binary-format raw-in-base64-out \ + --payload '{"count": 200}' /dev/stdout +``` -bin/kafka-share-groups.sh --bootstrap-server localhost:9092 \ - --describe --group kqd-queue-group- -# Shows per-partition lag with multiple pollers assigned +**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 | 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] ``` -**Confirm share groups are enabled:** +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 -bin/kafka-features.sh --bootstrap-server localhost:9092 describe | grep share -# Should show: FinalizedVersionLevel: 1 +# 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 +# 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 From 057435a3f81b992990bf0643d939a59517eaaf57 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Wed, 16 Sep 2026 15:19:54 -0400 Subject: [PATCH 03/28] refactor: remove DynamoDB from pattern, drop sleep from worker handler --- .../src/worker/handler.py | 5 --- .../stacks/3-app.yaml | 32 +++---------------- 2 files changed, 5 insertions(+), 32 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/src/worker/handler.py b/smk-lambda-queue-mode-python-sam/src/worker/handler.py index 820194979..cda6e8e6d 100644 --- a/smk-lambda-queue-mode-python-sam/src/worker/handler.py +++ b/smk-lambda-queue-mode-python-sam/src/worker/handler.py @@ -3,7 +3,6 @@ import base64 import json import logging -import time logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -31,10 +30,6 @@ def lambda_handler(event, context): 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)", diff --git a/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml b/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml index 69534b1db..744e5a178 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml @@ -1,8 +1,8 @@ AWSTemplateFormatVersion: '2010-09-09' Transform: AWS::Serverless-2016-10-31 Description: >- - KQD Stack 3 - Worker Lambda, Producer Lambda, DynamoDB idempotency table, - and SQS DLQ for the Kafka Queue mode (KIP-932) pattern. + KQD 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 kqd-network stack (set UseExistingInfra=false) - Bring-your-own Kafka: provide BootstrapServers, VpcSubnetIds, VpcSecurityGroupId @@ -46,25 +46,12 @@ Parameters: Conditions: ImportInfra: !Equals [!Ref UseExistingInfra, "false"] - UseProvidedInfra: !Equals [!Ref UseExistingInfra, "true"] Resources: - IdempotencyTable: - Type: AWS::DynamoDB::Table - Properties: - TableName: !Sub ${AWS::StackName}-idempotency - BillingMode: PAY_PER_REQUEST - AttributeDefinitions: - - AttributeName: id - AttributeType: S - KeySchema: - - AttributeName: id - KeyType: HASH - TimeToLiveSpecification: - AttributeName: expiration - Enabled: true - + # 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: @@ -82,19 +69,10 @@ Resources: Handler: handler.lambda_handler MemorySize: 512 Timeout: 25 - Environment: - Variables: - IDEMPOTENCY_TABLE: !Ref IdempotencyTable Policies: - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole - Version: '2012-10-17' Statement: - - Effect: Allow - Action: - - dynamodb:PutItem - - dynamodb:GetItem - - dynamodb:DeleteItem - Resource: !GetAtt IdempotencyTable.Arn - Effect: Allow Action: sqs:SendMessage Resource: !GetAtt TaskWorkerDLQ.Arn From af0a2d36bc43a1effb7ff24d9f30cf32088fb8e8 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Wed, 16 Sep 2026 21:01:31 -0400 Subject: [PATCH 04/28] fix: Add time.sleep(0.5) to worker for concurrent poller visibility Keeps records inflight long enough to observe multiple pollers processing the same partition simultaneously during scaling tests. --- smk-lambda-queue-mode-python-sam/src/worker/handler.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/smk-lambda-queue-mode-python-sam/src/worker/handler.py b/smk-lambda-queue-mode-python-sam/src/worker/handler.py index cda6e8e6d..820194979 100644 --- a/smk-lambda-queue-mode-python-sam/src/worker/handler.py +++ b/smk-lambda-queue-mode-python-sam/src/worker/handler.py @@ -3,6 +3,7 @@ import base64 import json import logging +import time logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -30,6 +31,10 @@ def lambda_handler(event, context): 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)", From e16bfc56e0eeaf8445dfe81a99f7d117955d4667 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Wed, 16 Sep 2026 21:06:19 -0400 Subject: [PATCH 05/28] fix: Fix Python string escaping bug in create-esm.sh output Unterminated string literal in --payload print statement caused the script to exit with SyntaxError after ESM creation succeeded. Tested end-to-end against us-west-2. --- smk-lambda-queue-mode-python-sam/scripts/create-esm.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh b/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh index ca941fe66..3d397530d 100755 --- a/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh +++ b/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh @@ -130,7 +130,7 @@ print(' aws lambda invoke \\\\') print(' --function-name kqd-app-producer \\\\') print(' --region $REGION --profile $PROFILE \\\\') print(' --cli-binary-format raw-in-base64-out \\\\') -print(' --payload \\'\\'{\"count\": 20}\\'' /dev/stdout') +print(' --payload \'{"count": 20}\' /dev/stdout') print() print('Watch logs:') print(' aws logs tail /aws/lambda/kqd-app-worker --follow \\\\') From 3dc07947cadd22d2987ee4720328ee81ee5c21d4 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Wed, 16 Sep 2026 21:17:25 -0400 Subject: [PATCH 06/28] feat: Accept topic override in producer event payload --- .../src/producer/producer.py | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/src/producer/producer.py b/smk-lambda-queue-mode-python-sam/src/producer/producer.py index da24a1dda..dcb93a717 100644 --- a/smk-lambda-queue-mode-python-sam/src/producer/producer.py +++ b/smk-lambda-queue-mode-python-sam/src/producer/producer.py @@ -3,6 +3,7 @@ PLAINTEXT, no auth. Every 7th item (taskIndex % 7 == 0) has shouldFail=True to demonstrate the RELEASE/retry path. """ + import json import logging import os @@ -23,15 +24,15 @@ def _config(): return {"bootstrap.servers": BOOTSTRAP_SERVERS} -def _ensure_topic(admin): - nt = NewTopic(TOPIC, num_partitions=3, replication_factor=1) - for topic, future in admin.create_topics([nt]).items(): +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", topic) + logger.info("Created topic %s", t) except KafkaException as e: if "already exists" in str(e).lower(): - logger.info("Topic %s already exists", topic) + logger.info("Topic %s already exists", t) else: raise @@ -39,8 +40,9 @@ def _ensure_topic(admin): 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())) + _ensure_topic(AdminClient(_config()), topic) producer = Producer(_config()) failures = 0 @@ -59,7 +61,7 @@ def on_delivery(err, _): "shouldFail": i % 7 == 0, } producer.produce( - TOPIC, + topic, key=item["jobId"].encode(), value=json.dumps(item).encode(), on_delivery=on_delivery, @@ -67,6 +69,6 @@ def on_delivery(err, _): producer.poll(0) producer.flush() - result = {"topic": TOPIC, "produced": count, "failures": failures} + result = {"topic": topic, "produced": count, "failures": failures} logger.info("Producer done: %s", json.dumps(result)) return result From fee92aa673c1453bf81c263c74753c66669fa17e Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Wed, 16 Sep 2026 21:31:58 -0400 Subject: [PATCH 07/28] refactor: Rename kqd- prefix to kafka-queue- throughout pattern Replaces all kqd- prefixes with kafka-queue- in stack names, CFN exports, resource names, scripts, and README so the pattern is self-contained and doesn't conflict with the kqd demo account. --- smk-lambda-queue-mode-python-sam/README.md | 24 +++++++------- .../events/kafka-event.json | 10 +++--- .../example-pattern.json | 2 +- .../scripts/create-esm.sh | 16 +++++----- .../src/producer/producer.py | 2 +- .../stacks/1-network.yaml | 32 +++++++++---------- .../stacks/2-broker.yaml | 10 +++--- .../stacks/3-app.yaml | 32 +++++++++---------- .../stacks/4-observability.yaml | 20 ++++++------ 9 files changed, 74 insertions(+), 74 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index e6d3b5b32..d38ccb462 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -58,7 +58,7 @@ cd serverless-patterns/smk-lambda-queue-mode-python-sam ```bash aws cloudformation deploy \ - --stack-name kqd-network \ + --stack-name kafka-queue-network \ --template-file stacks/1-network.yaml \ --region ``` @@ -67,7 +67,7 @@ aws cloudformation deploy \ ```bash aws cloudformation deploy \ - --stack-name kqd-broker \ + --stack-name kafka-queue-broker \ --template-file stacks/2-broker.yaml \ --capabilities CAPABILITY_IAM \ --region @@ -80,7 +80,7 @@ This provisions a t3.medium EC2 instance running Apache Kafka 4.2.x in KRaft mod ```bash sam build --template stacks/3-app.yaml sam deploy \ - --stack-name kqd-app \ + --stack-name kafka-queue-app \ --template-file .aws-sam/build/template.yaml \ --capabilities CAPABILITY_IAM \ --resolve-s3 \ @@ -91,7 +91,7 @@ sam deploy \ ```bash aws cloudformation deploy \ - --stack-name kqd-observability \ + --stack-name kafka-queue-observability \ --template-file stacks/4-observability.yaml \ --region ``` @@ -105,7 +105,7 @@ Skip Steps 1 and 2. Provide your Kafka bootstrap servers, VPC subnet IDs, and se ```bash sam build --template stacks/3-app.yaml sam deploy \ - --stack-name kqd-app \ + --stack-name kafka-queue-app \ --template-file .aws-sam/build/template.yaml \ --capabilities CAPABILITY_IAM \ --resolve-s3 \ @@ -152,7 +152,7 @@ Wait ~60 seconds for the ESM to reach `State: Enabled`. ```bash aws lambda invoke \ - --function-name kqd-app-producer \ + --function-name kafka-queue-app-producer \ --region \ --cli-binary-format raw-in-base64-out \ --payload '{"count": 20}' /dev/stdout @@ -163,7 +163,7 @@ Every 7th record (`taskIndex % 7 == 0`) has `shouldFail: true` to demonstrate th **Watch the worker Lambda logs:** ```bash -aws logs tail /aws/lambda/kqd-app-worker \ +aws logs tail /aws/lambda/kafka-queue-app-worker \ --follow \ --filter-pattern KAFKA_RECORD \ --region @@ -189,7 +189,7 @@ The key differentiator of Queue mode is that pollers exceed the partition count. ```bash aws lambda invoke \ - --function-name kqd-app-producer \ + --function-name kafka-queue-app-producer \ --region \ --cli-binary-format raw-in-base64-out \ --payload '{"count": 200}' /dev/stdout @@ -268,14 +268,14 @@ Delete stacks in reverse order: aws lambda delete-event-source-mapping --uuid --region # 2. Delete application stacks -aws cloudformation delete-stack --stack-name kqd-observability --region -aws cloudformation delete-stack --stack-name kqd-app --region +aws cloudformation delete-stack --stack-name kafka-queue-observability --region +aws cloudformation delete-stack --stack-name kafka-queue-app --region # 3. Delete broker (if deployed) -aws cloudformation delete-stack --stack-name kqd-broker --region +aws cloudformation delete-stack --stack-name kafka-queue-broker --region # 4. Delete network (if deployed) -aws cloudformation delete-stack --stack-name kqd-network --region +aws cloudformation delete-stack --stack-name kafka-queue-network --region ``` --- diff --git a/smk-lambda-queue-mode-python-sam/events/kafka-event.json b/smk-lambda-queue-mode-python-sam/events/kafka-event.json index b3f0d61fe..4b38bd8f7 100644 --- a/smk-lambda-queue-mode-python-sam/events/kafka-event.json +++ b/smk-lambda-queue-mode-python-sam/events/kafka-event.json @@ -2,9 +2,9 @@ "eventSource": "aws:kafka", "bootstrapServers": "10.0.0.43:9092", "records": { - "kqd-task-worker-0": [ + "kafka-queue-task-worker-0": [ { - "topic": "kqd-task-worker", + "topic": "kafka-queue-task-worker", "partition": 0, "offset": 0, "timestamp": 1726000000000, @@ -14,7 +14,7 @@ "headers": [] }, { - "topic": "kqd-task-worker", + "topic": "kafka-queue-task-worker", "partition": 0, "offset": 1, "timestamp": 1726000001000, @@ -24,9 +24,9 @@ "headers": [] } ], - "kqd-task-worker-1": [ + "kafka-queue-task-worker-1": [ { - "topic": "kqd-task-worker", + "topic": "kafka-queue-task-worker", "partition": 1, "offset": 0, "timestamp": 1726000000500, diff --git a/smk-lambda-queue-mode-python-sam/example-pattern.json b/smk-lambda-queue-mode-python-sam/example-pattern.json index 0f318dcb7..156ee4018 100644 --- a/smk-lambda-queue-mode-python-sam/example-pattern.json +++ b/smk-lambda-queue-mode-python-sam/example-pattern.json @@ -56,7 +56,7 @@ }, "cleanup": { "text": [ - "Delete stacks in reverse order: aws cloudformation delete-stack --stack-name kqd-esm, then kqd-app, kqd-broker, kqd-network." + "Delete stacks in reverse order: aws cloudformation delete-stack --stack-name kafka-queue-esm, then kafka-queue-app, kafka-queue-broker, kafka-queue-network." ] }, "authors": [ diff --git a/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh b/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh index 3d397530d..20fdd1396 100755 --- a/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh +++ b/smk-lambda-queue-mode-python-sam/scripts/create-esm.sh @@ -8,7 +8,7 @@ # Prerequisites: # - aws CLI configured with appropriate credentials # - curl >= 7.75 (for --aws-sigv4 support) -# - Stacks kqd-network, kqd-broker, kqd-app must be deployed +# - Stacks kafka-queue-network, kafka-queue-broker, kafka-queue-app must be deployed # # Usage: # ./scripts/create-esm.sh @@ -19,11 +19,11 @@ set -euo pipefail # ── Defaults ───────────────────────────────────────────────── REGION="${AWS_DEFAULT_REGION:-us-east-1}" PROFILE="${AWS_PROFILE:-default}" -APP_STACK="kqd-app" -BROKER_STACK="kqd-broker" -NETWORK_STACK="kqd-network" -CONSUMER_GROUP_ID="kqd-queue-group-$(date +%s)" -TOPIC="kqd-task-worker" +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 @@ -127,13 +127,13 @@ print() print('Wait ~60s for State to reach Enabled, then produce records:') print() print(' aws lambda invoke \\\\') -print(' --function-name kqd-app-producer \\\\') +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/kqd-app-worker --follow \\\\') +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/src/producer/producer.py b/smk-lambda-queue-mode-python-sam/src/producer/producer.py index dcb93a717..f4b587df0 100644 --- a/smk-lambda-queue-mode-python-sam/src/producer/producer.py +++ b/smk-lambda-queue-mode-python-sam/src/producer/producer.py @@ -15,7 +15,7 @@ logger = logging.getLogger() logger.setLevel(logging.INFO) -TOPIC = os.environ.get("KAFKA_TOPIC", "kqd-task-worker") +TOPIC = os.environ.get("KAFKA_TOPIC", "kafka-queue-task-worker") BOOTSTRAP_SERVERS = os.environ.get("BOOTSTRAP_SERVERS", "") DEFAULT_COUNT = 50 diff --git a/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml index 2d1f893a9..033d808da 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml @@ -11,14 +11,14 @@ Resources: EnableDnsHostnames: true Tags: - Key: Name - Value: kqd-vpc + Value: kafka-queue-vpc InternetGateway: Type: AWS::EC2::InternetGateway Properties: Tags: - Key: Name - Value: kqd-igw + Value: kafka-queue-igw VPCGatewayAttachment: Type: AWS::EC2::VPCGatewayAttachment @@ -35,7 +35,7 @@ Resources: MapPublicIpOnLaunch: true Tags: - Key: Name - Value: kqd-public + Value: kafka-queue-public PublicRouteTable: Type: AWS::EC2::RouteTable @@ -43,7 +43,7 @@ Resources: VpcId: !Ref VPC Tags: - Key: Name - Value: kqd-public-rt + Value: kafka-queue-public-rt PublicRoute: Type: AWS::EC2::Route @@ -67,7 +67,7 @@ Resources: AvailabilityZone: !Select [0, !GetAZs ""] Tags: - Key: Name - Value: kqd-private-a + Value: kafka-queue-private-a PrivateSubnetB: Type: AWS::EC2::Subnet @@ -77,7 +77,7 @@ Resources: AvailabilityZone: !Select [1, !GetAZs ""] Tags: - Key: Name - Value: kqd-private-b + Value: kafka-queue-private-b PrivateSubnetC: Type: AWS::EC2::Subnet @@ -87,7 +87,7 @@ Resources: AvailabilityZone: !Select [2, !GetAZs ""] Tags: - Key: Name - Value: kqd-private-c + Value: kafka-queue-private-c BrokerSecurityGroup: Type: AWS::EC2::SecurityGroup @@ -102,7 +102,7 @@ Resources: Description: Kafka PLAINTEXT from VPC Tags: - Key: Name - Value: kqd-broker-sg + Value: kafka-queue-broker-sg LambdaSecurityGroup: Type: AWS::EC2::SecurityGroup @@ -122,7 +122,7 @@ Resources: Description: HTTPS for AWS APIs and VPC endpoints Tags: - Key: Name - Value: kqd-lambda-sg + Value: kafka-queue-lambda-sg # Allow inbound 443 from VPC (VPC endpoint ENIs need this) LambdaSGIngressVPC: @@ -159,34 +159,34 @@ Outputs: VpcId: Value: !Ref VPC Export: - Name: kqd-VpcId + Name: kafka-queue-VpcId PublicSubnetId: Value: !Ref PublicSubnet Export: - Name: kqd-PublicSubnetId + Name: kafka-queue-PublicSubnetId PrivateSubnetA: Value: !Ref PrivateSubnetA Export: - Name: kqd-PrivateSubnetA + Name: kafka-queue-PrivateSubnetA PrivateSubnetB: Value: !Ref PrivateSubnetB Export: - Name: kqd-PrivateSubnetB + Name: kafka-queue-PrivateSubnetB PrivateSubnetC: Value: !Ref PrivateSubnetC Export: - Name: kqd-PrivateSubnetC + Name: kafka-queue-PrivateSubnetC BrokerSecurityGroupId: Value: !Ref BrokerSecurityGroup Export: - Name: kqd-BrokerSecurityGroupId + Name: kafka-queue-BrokerSecurityGroupId LambdaSecurityGroupId: Value: !Ref LambdaSecurityGroup Export: - Name: kqd-LambdaSecurityGroupId + 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 index becb9b8ef..c491716f4 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml @@ -54,13 +54,13 @@ Resources: Properties: InstanceType: !Ref InstanceType ImageId: !Ref LatestAL2023AmiId - SubnetId: !ImportValue kqd-PublicSubnetId + SubnetId: !ImportValue kafka-queue-PublicSubnetId SecurityGroupIds: - - !ImportValue kqd-BrokerSecurityGroupId + - !ImportValue kafka-queue-BrokerSecurityGroupId IamInstanceProfile: !Ref KafkaInstanceProfile Tags: - Key: Name - Value: kqd-kafka-broker + Value: kafka-queue-kafka-broker UserData: Fn::Base64: !Sub | #!/bin/bash @@ -154,9 +154,9 @@ Outputs: BootstrapServers: Value: !Sub "${KafkaInstance.PrivateIp}:9092" Export: - Name: kqd-BootstrapServers + Name: kafka-queue-BootstrapServers BrokerInstanceId: Value: !Ref KafkaInstance Export: - Name: kqd-BrokerInstanceId + 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 index 744e5a178..c668df175 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml @@ -4,14 +4,14 @@ Description: >- KQD 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 kqd-network stack (set UseExistingInfra=false) + - 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: kqd-task-worker + Default: kafka-queue-task-worker Description: Kafka topic name the worker consumes and the producer publishes to. BootstrapServers: @@ -19,21 +19,21 @@ Parameters: Default: "" Description: >- Kafka bootstrap servers (e.g. 10.0.0.43:9092). - Leave blank to import from kqd-broker stack output. + 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 kqd-network stack output. + 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 kqd-network stack output. + Leave blank to import from kafka-queue-network stack output. UseExistingInfra: Type: String @@ -41,8 +41,8 @@ Parameters: AllowedValues: ["true", "false"] Description: >- Set to true when providing your own BootstrapServers, VpcSubnetIds, - and VpcSecurityGroupId (skips CloudFormation imports from kqd-network - and kqd-broker stacks). + and VpcSecurityGroupId (skips CloudFormation imports from kafka-queue-network + and kafka-queue-broker stacks). Conditions: ImportInfra: !Equals [!Ref UseExistingInfra, "false"] @@ -102,21 +102,21 @@ Resources: - ImportInfra - !Join - "," - - - !ImportValue kqd-PrivateSubnetA - - !ImportValue kqd-PrivateSubnetB - - !ImportValue kqd-PrivateSubnetC + - - !ImportValue kafka-queue-PrivateSubnetA + - !ImportValue kafka-queue-PrivateSubnetB + - !ImportValue kafka-queue-PrivateSubnetC - !Ref VpcSubnetIds SecurityGroupIds: - !If - ImportInfra - - !ImportValue kqd-LambdaSecurityGroupId + - !ImportValue kafka-queue-LambdaSecurityGroupId - !Ref VpcSecurityGroupId Environment: Variables: KAFKA_TOPIC: !Ref KafkaTopic BOOTSTRAP_SERVERS: !If - ImportInfra - - !ImportValue kqd-BootstrapServers + - !ImportValue kafka-queue-BootstrapServers - !Ref BootstrapServers Policies: - arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole @@ -125,19 +125,19 @@ Outputs: WorkerFunctionArn: Value: !GetAtt WorkerFunction.Arn Export: - Name: kqd-WorkerFunctionArn + Name: kafka-queue-WorkerFunctionArn WorkerFunctionName: Value: !Ref WorkerFunction Export: - Name: kqd-WorkerFunctionName + Name: kafka-queue-WorkerFunctionName TaskWorkerDLQArn: Value: !GetAtt TaskWorkerDLQ.Arn Export: - Name: kqd-TaskWorkerDLQArn + Name: kafka-queue-TaskWorkerDLQArn TaskWorkerDLQUrl: Value: !Ref TaskWorkerDLQ Export: - Name: kqd-TaskWorkerDLQUrl + 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 index 2dfec3ffa..142db0e58 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml @@ -6,18 +6,18 @@ Resources: AlarmTopic: Type: AWS::SNS::Topic Properties: - TopicName: kqd-alarms + TopicName: kafka-queue-alarms DlqDeliveryAlarm: Type: AWS::CloudWatch::Alarm Properties: - AlarmName: kqd-dlq-delivery + AlarmName: kafka-queue-dlq-delivery AlarmDescription: Records routed to DLQ (delivery attempts exhausted) Namespace: AWS/Lambda MetricName: OnFailureDestinationDeliveredEventCount Dimensions: - Name: FunctionName - Value: !ImportValue kqd-WorkerFunctionName + Value: !ImportValue kafka-queue-WorkerFunctionName Statistic: Sum Period: 60 EvaluationPeriods: 1 @@ -30,13 +30,13 @@ Resources: LagGrowthAlarm: Type: AWS::CloudWatch::Alarm Properties: - AlarmName: kqd-lag-growth + AlarmName: kafka-queue-lag-growth AlarmDescription: Share group offset lag is growing Namespace: AWS/Lambda MetricName: MaxOffsetLag Dimensions: - Name: FunctionName - Value: !ImportValue kqd-WorkerFunctionName + Value: !ImportValue kafka-queue-WorkerFunctionName Statistic: Maximum Period: 300 EvaluationPeriods: 2 @@ -49,13 +49,13 @@ Resources: PollerErrorAlarm: Type: AWS::CloudWatch::Alarm Properties: - AlarmName: kqd-poller-errors + AlarmName: kafka-queue-poller-errors AlarmDescription: ESM reported polling errors Namespace: AWS/Lambda MetricName: PollingErrorCount Dimensions: - Name: FunctionName - Value: !ImportValue kqd-WorkerFunctionName + Value: !ImportValue kafka-queue-WorkerFunctionName Statistic: Sum Period: 60 EvaluationPeriods: 1 @@ -68,7 +68,7 @@ Resources: KafkaQueueDashboard: Type: AWS::CloudWatch::Dashboard Properties: - DashboardName: kqd-KafkaQueue + DashboardName: kafka-queue-KafkaQueue DashboardBody: !Sub - | {"widgets":[ @@ -77,10 +77,10 @@ Resources: {"type":"metric","x":0,"y":6,"width":12,"height":6,"properties":{"title":"Event Counts","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","PolledEventCount","FunctionName","${Fn}"],["AWS/Lambda","InvokedEventCount","FunctionName","${Fn}"],["AWS/Lambda","AcknowledgedEventCount","FunctionName","${Fn}"],["AWS/Lambda","OnFailureDestinationDeliveredEventCount","FunctionName","${Fn}"]]}}, {"type":"metric","x":12,"y":6,"width":12,"height":6,"properties":{"title":"Errors","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","PollingErrorCount","FunctionName","${Fn}"],["AWS/Lambda","InvokeErrorCount","FunctionName","${Fn}"]]}} ]} - - Fn: !ImportValue kqd-WorkerFunctionName + - Fn: !ImportValue kafka-queue-WorkerFunctionName Outputs: AlarmTopicArn: Value: !Ref AlarmTopic Export: - Name: kqd-AlarmTopicArn + Name: kafka-queue-AlarmTopicArn From a7419d58aa8a8d5768fa75a8303ffcb8303d1bb1 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Wed, 16 Sep 2026 21:39:34 -0400 Subject: [PATCH 08/28] =?UTF-8?q?fix:=20Fix=20observability=20stack=20?= =?UTF-8?q?=E2=80=94=20correct=20dimension=20and=20metric=20names?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Dimension: FunctionName -> EventSourceMappingUUID - MaxOffsetLag -> MaxShareGroupLag - SumOffsetLag -> SumShareGroupLag - Remove unused SNS topic and AcknowledgedEventCount - Add ESMUuid parameter - Verified all metrics populate in us-west-2 kqd account --- .../stacks/4-observability.yaml | 52 +++++++------------ 1 file changed, 20 insertions(+), 32 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml b/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml index 142db0e58..41dd53d6a 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/4-observability.yaml @@ -1,12 +1,12 @@ AWSTemplateFormatVersion: '2010-09-09' -Description: KQD Stack 4 - CloudWatch alarms and dashboard for Queue mode ESM +Description: Stack 4 - CloudWatch alarms and dashboard for Queue mode ESM -Resources: +Parameters: + ESMUuid: + Type: String + Description: UUID of the Queue mode Event Source Mapping (from create-esm.sh output) - AlarmTopic: - Type: AWS::SNS::Topic - Properties: - TopicName: kafka-queue-alarms +Resources: DlqDeliveryAlarm: Type: AWS::CloudWatch::Alarm @@ -16,35 +16,31 @@ Resources: Namespace: AWS/Lambda MetricName: OnFailureDestinationDeliveredEventCount Dimensions: - - Name: FunctionName - Value: !ImportValue kafka-queue-WorkerFunctionName + - Name: EventSourceMappingUUID + Value: !Ref ESMUuid Statistic: Sum Period: 60 EvaluationPeriods: 1 Threshold: 1 ComparisonOperator: GreaterThanOrEqualToThreshold TreatMissingData: notBreaching - AlarmActions: - - !Ref AlarmTopic LagGrowthAlarm: Type: AWS::CloudWatch::Alarm Properties: AlarmName: kafka-queue-lag-growth - AlarmDescription: Share group offset lag is growing + AlarmDescription: Share group lag is growing Namespace: AWS/Lambda - MetricName: MaxOffsetLag + MetricName: MaxShareGroupLag Dimensions: - - Name: FunctionName - Value: !ImportValue kafka-queue-WorkerFunctionName + - Name: EventSourceMappingUUID + Value: !Ref ESMUuid Statistic: Maximum Period: 300 EvaluationPeriods: 2 Threshold: 1000 ComparisonOperator: GreaterThanThreshold TreatMissingData: notBreaching - AlarmActions: - - !Ref AlarmTopic PollerErrorAlarm: Type: AWS::CloudWatch::Alarm @@ -54,33 +50,25 @@ Resources: Namespace: AWS/Lambda MetricName: PollingErrorCount Dimensions: - - Name: FunctionName - Value: !ImportValue kafka-queue-WorkerFunctionName + - Name: EventSourceMappingUUID + Value: !Ref ESMUuid Statistic: Sum Period: 60 EvaluationPeriods: 1 Threshold: 1 ComparisonOperator: GreaterThanOrEqualToThreshold TreatMissingData: notBreaching - AlarmActions: - - !Ref AlarmTopic KafkaQueueDashboard: Type: AWS::CloudWatch::Dashboard Properties: - DashboardName: kafka-queue-KafkaQueue + DashboardName: kafka-queue-dashboard DashboardBody: !Sub - | {"widgets":[ - {"type":"metric","x":0,"y":0,"width":12,"height":6,"properties":{"title":"Offset Lag","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","SumOffsetLag","FunctionName","${Fn}"],["AWS/Lambda","MaxOffsetLag","FunctionName","${Fn}"]]}}, - {"type":"metric","x":12,"y":0,"width":12,"height":6,"properties":{"title":"Pollers","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","ProvisionedPollers","FunctionName","${Fn}"]]}}, - {"type":"metric","x":0,"y":6,"width":12,"height":6,"properties":{"title":"Event Counts","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","PolledEventCount","FunctionName","${Fn}"],["AWS/Lambda","InvokedEventCount","FunctionName","${Fn}"],["AWS/Lambda","AcknowledgedEventCount","FunctionName","${Fn}"],["AWS/Lambda","OnFailureDestinationDeliveredEventCount","FunctionName","${Fn}"]]}}, - {"type":"metric","x":12,"y":6,"width":12,"height":6,"properties":{"title":"Errors","region":"${AWS::Region}","view":"timeSeries","metrics":[["AWS/Lambda","PollingErrorCount","FunctionName","${Fn}"],["AWS/Lambda","InvokeErrorCount","FunctionName","${Fn}"]]}} + {"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}"]]}} ]} - - Fn: !ImportValue kafka-queue-WorkerFunctionName - -Outputs: - AlarmTopicArn: - Value: !Ref AlarmTopic - Export: - Name: kafka-queue-AlarmTopicArn + - ESM: !Ref ESMUuid From 4719f162b8d4d2cc41ea03247055ef4168443d91 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 10:05:50 -0400 Subject: [PATCH 09/28] fix: Add VPC endpoint setup script and fix README deployment flow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add scripts/setup-vpc-endpoints.sh — idempotent creation of Lambda, STS, and SQS interface endpoints. Supports Path A (resolves from kafka-queue-network stack) and Path B/C (accepts --vpc-id, --subnet-ids, --security-group-id flags for BYO VPC). - README Path A: add Step 4 (VPC endpoints) and Step 5 (ESM), move observability to Step 6 with required ESMUuid parameter. - README Path B: add VPC endpoint step with BYO VPC flags. - README Path C: clarify setup-vpc-endpoints.sh usage. - 1-network.yaml: update comment to document all 3 required endpoints including SQS, with explanation of silent failure if SQS is missing. --- smk-lambda-queue-mode-python-sam/README.md | 55 ++++++--- .../scripts/setup-vpc-endpoints.sh | 108 ++++++++++++++++++ .../stacks/1-network.yaml | 20 ++-- 3 files changed, 161 insertions(+), 22 deletions(-) create mode 100755 smk-lambda-queue-mode-python-sam/scripts/setup-vpc-endpoints.sh diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index d38ccb462..3fe1a8746 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -87,18 +87,39 @@ sam deploy \ --region ``` -**Step 4: Deploy observability (optional)** +**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 --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 --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= \ --region ``` ---- - -### Path B: Bring your own Kafka cluster +This creates a CloudWatch dashboard (`kafka-queue-dashboard`) and alarms for share group lag, DLQ delivery, and poller errors. Skip Steps 1 and 2. Provide your Kafka bootstrap servers, VPC subnet IDs, and security group at deploy time: @@ -123,26 +144,32 @@ Your Kafka cluster must: - 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 ---- - -### Path C: Bring your own Kafka cluster and VPC +**Create VPC endpoints** -Same as Path B — `UseExistingInfra=true` handles both cases. +If your VPC does not already have `lambda`, `sts`, and `sqs` interface endpoints, create them: ---- +```bash +./scripts/setup-vpc-endpoints.sh --region --profile \ + --vpc-id \ + --subnet-ids "subnet-aaa111,subnet-bbb222,subnet-ccc333" \ + --security-group-id +``` -## Create the Queue mode ESM +If your VPC already has these endpoints, skip this step. -After deploying the application stack, create the Event Source Mapping with `ConsumptionMode: Queue`: +**Create the Queue mode ESM** ```bash -chmod +x scripts/create-esm.sh ./scripts/create-esm.sh --region --profile ``` -Wait ~60 seconds for the ESM to reach `State: Enabled`. +--- -> **Note:** The script uses `curl --aws-sigv4` to call the Lambda API directly because `ConsumptionMode: Queue` is not yet in the SAM or AWS CLI service model. Update to the latest AWS CLI or SAM when this field becomes available to use standard tooling. +### 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. + +--- --- 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 --profile +# +# Usage (Path B/C — BYO VPC, provide values directly): +# ./scripts/setup-vpc-endpoints.sh --region --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/stacks/1-network.yaml b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml index 033d808da..b6c3583d4 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml @@ -146,14 +146,18 @@ Resources: SourceSecurityGroupId: !Ref LambdaSecurityGroup Description: Self - Hyperplane ENI to VPC endpoint ENI - # VPC Interface Endpoint for Lambda (allows ESM pollers to invoke Lambda) - # NOTE: Created separately via CLI if this stack is deployed fresh: - # aws ec2 create-vpc-endpoint --vpc-endpoint-type Interface \ - # --vpc-id --service-name com.amazonaws..lambda \ - # --subnet-ids --security-group-ids \ - # --private-dns-enabled --region - # Same for com.amazonaws..sts - # These endpoints are not managed here to avoid conflict on update. + # 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: From 39c812ba8e760a82b3974cac19695bf466873d2b Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 11:25:13 -0400 Subject: [PATCH 10/28] fix: Clean up remaining pattern issues - Remove DynamoDB from architecture diagram and costs (table removed) - Fix 'KQD Stack N' descriptions in stacks 1, 2, 3 - Add topic override example to Testing section - Add CloudWatch dashboard section with metric descriptions - Add VPC endpoint cleanup to Cleanup section --- smk-lambda-queue-mode-python-sam/README.md | 37 ++++++++++++++++--- .../stacks/1-network.yaml | 2 +- .../stacks/2-broker.yaml | 2 +- .../stacks/3-app.yaml | 2 +- 4 files changed, 35 insertions(+), 8 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index 3fe1a8746..f9ee54255 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -12,7 +12,6 @@ Learn more about this pattern at Serverless Land: https://serverlessland.com/pat graph LR Producer["Producer\nLambda"] -->|publish| Kafka["Apache Kafka\n4.2+ Cluster"] Kafka -->|Queue mode ESM\nConsumptionMode: Queue| Worker["Worker\nLambda"] - Worker --> DDB["DynamoDB\nIdempotency"] Worker --> SQS["SQS\nDLQ"] Worker --> CW["CloudWatch\nMetrics"] ``` @@ -29,7 +28,7 @@ graph LR ## Costs -This pattern uses EC2 (t3.medium), Lambda, DynamoDB, 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. +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. --- @@ -185,6 +184,16 @@ aws lambda invoke \ --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 \ + --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:** @@ -198,6 +207,16 @@ aws logs tail /aws/lambda/kafka-queue-app-worker \ 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 @@ -294,14 +313,22 @@ Delete stacks in reverse order: # 1. Delete the ESM first (get UUID from create-esm.sh output or console) aws lambda delete-event-source-mapping --uuid --region -# 2. Delete application stacks +# 2. Delete VPC endpoints +aws ec2 describe-vpc-endpoints \ + --filters "Name=vpc-id,Values=" \ + "Name=service-name,Values=com.amazonaws..lambda,com.amazonaws..sts,com.amazonaws..sqs" \ + --query 'VpcEndpoints[*].VpcEndpointId' --output text --region +# Then delete each endpoint: +aws ec2 delete-vpc-endpoints --vpc-endpoint-ids --region + +# 3. Delete application stacks aws cloudformation delete-stack --stack-name kafka-queue-observability --region aws cloudformation delete-stack --stack-name kafka-queue-app --region -# 3. Delete broker (if deployed) +# 4. Delete broker (if deployed) aws cloudformation delete-stack --stack-name kafka-queue-broker --region -# 4. Delete network (if deployed) +# 5. Delete network (if deployed) aws cloudformation delete-stack --stack-name kafka-queue-network --region ``` diff --git a/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml index b6c3583d4..ba4c12b0a 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/1-network.yaml @@ -1,5 +1,5 @@ AWSTemplateFormatVersion: '2010-09-09' -Description: KQD Stack 1 - Networking (VPC, subnets, IGW, security groups) +Description: Stack 1 - Networking (VPC, subnets, IGW, security groups) Resources: diff --git a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml index c491716f4..880a07045 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml @@ -1,5 +1,5 @@ AWSTemplateFormatVersion: '2010-09-09' -Description: KQD Stack 2 - Self-managed Kafka 4.2.x broker on EC2 (KRaft, PLAINTEXT) +Description: Stack 2 - Self-managed Kafka 4.2.x broker on EC2 (KRaft, PLAINTEXT) Parameters: KafkaVersion: diff --git a/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml b/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml index c668df175..a2e07d0f2 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/3-app.yaml @@ -1,7 +1,7 @@ AWSTemplateFormatVersion: '2010-09-09' Transform: AWS::Serverless-2016-10-31 Description: >- - KQD Stack 3 - Worker Lambda, Producer Lambda, and SQS DLQ + 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) From ab38e10ca941a058cadd0419dc68f4cbcbda1339 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 12:42:27 -0400 Subject: [PATCH 11/28] fix: Add sleep and more retries before Kafka download in UserData --- smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml index 880a07045..61fc87a27 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml @@ -81,7 +81,9 @@ Resources: KAFKA_DIR=/opt/kafka mkdir -p $KAFKA_DIR - wget -q --timeout=180 --tries=3 \ + # Wait for network routing to settle after instance launch + sleep 15 + wget -q --timeout=180 --tries=5 \ "https://archive.apache.org/dist/kafka/${KafkaVersion}/kafka_2.13-${KafkaVersion}.tgz" \ -O /tmp/kafka.tgz tar -xzf /tmp/kafka.tgz -C $KAFKA_DIR --strip-components=1 From 52232408341f1d40225464db264077deb719a00e Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 13:02:54 -0400 Subject: [PATCH 12/28] fix: Replace tee subshell with direct file redirect in UserData --- smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml index 61fc87a27..8c3ba21e4 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml @@ -65,7 +65,7 @@ Resources: Fn::Base64: !Sub | #!/bin/bash set -euo pipefail - exec > >(tee /var/log/kafka-setup.log) 2>&1 + exec >> /var/log/kafka-setup.log 2>&1 echo "=== Kafka ${KafkaVersion} setup $(date) ===" signal_cfn() { From 40679882606fb59bcf41440f48c7c1a29f7a99eb Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 14:40:26 -0400 Subject: [PATCH 13/28] fix: Increase broker CFN signal timeout to 30 min --- smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml index 8c3ba21e4..a5b6d054a 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml @@ -65,7 +65,7 @@ Resources: Fn::Base64: !Sub | #!/bin/bash set -euo pipefail - exec >> /var/log/kafka-setup.log 2>&1 + exec > >(tee /var/log/kafka-setup.log) 2>&1 echo "=== Kafka ${KafkaVersion} setup $(date) ===" signal_cfn() { @@ -81,8 +81,6 @@ Resources: KAFKA_DIR=/opt/kafka mkdir -p $KAFKA_DIR - # Wait for network routing to settle after instance launch - sleep 15 wget -q --timeout=180 --tries=5 \ "https://archive.apache.org/dist/kafka/${KafkaVersion}/kafka_2.13-${KafkaVersion}.tgz" \ -O /tmp/kafka.tgz @@ -146,7 +144,7 @@ Resources: CreationPolicy: ResourceSignal: - Timeout: PT20M + Timeout: PT30M # Dummy resource to ensure IGW is attached before EC2 launches VPCGatewayAttachmentDependency: From 9514e87d1136e7d15454577ba55a2fb9763d7ec4 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 14:40:59 -0400 Subject: [PATCH 14/28] fix: Increase broker CFN signal timeout to 40 min --- smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml index a5b6d054a..c7eb2e70a 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml @@ -144,7 +144,7 @@ Resources: CreationPolicy: ResourceSignal: - Timeout: PT30M + Timeout: PT40M # Dummy resource to ensure IGW is attached before EC2 launches VPCGatewayAttachmentDependency: From f3fb7cd90eaaeb8dec82f50dd8dc6bfd8711cfbc Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 15:19:58 -0400 Subject: [PATCH 15/28] refactor: Move Kafka install from UserData to setup-broker.sh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit UserData was fragile — silent exits on newer AL2023 kernels when combining exec > >(tee ...) with set -euo pipefail. The wget also takes 15-20 min which required a 40-min CFN timeout. New approach: - 2-broker.yaml: launches EC2 instance only (no UserData) - scripts/setup-broker.sh: installs Kafka via 6 sequential SSM send-command steps with progress reporting and error handling - README: updated all 3 paths to reference setup-broker.sh No SSH or bastion required — all steps run via SSM. --- smk-lambda-queue-mode-python-sam/README.md | 13 +- .../scripts/setup-broker.sh | 153 ++++++++++++++++++ .../stacks/2-broker.yaml | 107 ------------ 3 files changed, 164 insertions(+), 109 deletions(-) create mode 100755 smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index f9ee54255..c878c6b4d 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -72,7 +72,16 @@ aws cloudformation deploy \ --region ``` -This provisions a t3.medium EC2 instance running Apache Kafka 4.2.x in KRaft mode with share groups enabled. The stack signals CloudFormation when Kafka is ready (~10 minutes). +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 --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** @@ -120,7 +129,7 @@ aws cloudformation deploy \ This creates a CloudWatch dashboard (`kafka-queue-dashboard`) and alarms for share group lag, DLQ delivery, and poller errors. -Skip Steps 1 and 2. Provide your Kafka bootstrap servers, VPC subnet IDs, and security group at deploy time: +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 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..9ca41177b --- /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 --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=1800 --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" \ + 1800 + +# ── 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/stacks/2-broker.yaml b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml index c7eb2e70a..01d42a003 100644 --- a/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml +++ b/smk-lambda-queue-mode-python-sam/stacks/2-broker.yaml @@ -2,9 +2,6 @@ AWSTemplateFormatVersion: '2010-09-09' Description: Stack 2 - Self-managed Kafka 4.2.x broker on EC2 (KRaft, PLAINTEXT) Parameters: - KafkaVersion: - Type: String - Default: 4.2.0 InstanceType: Type: String Default: t3.medium @@ -26,21 +23,6 @@ Resources: Action: sts:AssumeRole ManagedPolicyArns: - arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore - - arn:aws:iam::aws:policy/AmazonSSMPatchAssociation - - # Automated patching: runs AWS-RunPatchBaseline every Sunday at 2am UTC - PatchAssociation: - Type: AWS::SSM::Association - Properties: - Name: AWS-RunPatchBaseline - Targets: - - Key: tag:aws:cloudformation:stack-name - Values: - - !Ref AWS::StackName - ScheduleExpression: cron(0 2 ? * SUN *) - Parameters: - Operation: - - Install KafkaInstanceProfile: Type: AWS::IAM::InstanceProfile @@ -50,7 +32,6 @@ Resources: KafkaInstance: Type: AWS::EC2::Instance - DependsOn: VPCGatewayAttachmentDependency Properties: InstanceType: !Ref InstanceType ImageId: !Ref LatestAL2023AmiId @@ -61,94 +42,6 @@ Resources: Tags: - Key: Name Value: kafka-queue-kafka-broker - UserData: - Fn::Base64: !Sub | - #!/bin/bash - set -euo pipefail - exec > >(tee /var/log/kafka-setup.log) 2>&1 - echo "=== Kafka ${KafkaVersion} setup $(date) ===" - - signal_cfn() { - /opt/aws/bin/cfn-signal -e $1 \ - --stack ${AWS::StackName} \ - --resource KafkaInstance \ - --region ${AWS::Region} || true - } - trap 'echo "ERROR line $LINENO"; signal_cfn 1' ERR - - dnf install -y java-21-amazon-corretto-headless - echo "Java installed." - - KAFKA_DIR=/opt/kafka - mkdir -p $KAFKA_DIR - wget -q --timeout=180 --tries=5 \ - "https://archive.apache.org/dist/kafka/${KafkaVersion}/kafka_2.13-${KafkaVersion}.tgz" \ - -O /tmp/kafka.tgz - tar -xzf /tmp/kafka.tgz -C $KAFKA_DIR --strip-components=1 - echo "Kafka extracted." - - INSTANCE_IP=$(curl -s http://169.254.169.254/latest/meta-data/local-ipv4) - mkdir -p /var/kafka-logs - - PROPS=/tmp/kraft-server.properties - 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" \ - > $PROPS - - CLUSTER_ID=$($KAFKA_DIR/bin/kafka-storage.sh random-uuid) - $KAFKA_DIR/bin/kafka-storage.sh format -t $CLUSTER_ID -c $PROPS - echo "KRaft storage formatted." - - export KAFKA_HEAP_OPTS="-Xmx512m -Xms256m" - nohup $KAFKA_DIR/bin/kafka-server-start.sh $PROPS \ - > /var/log/kafka.log 2>&1 & - echo "Kafka started (PID $!)" - - echo "Waiting for broker..." - for i in $(seq 1 24); do - $KAFKA_DIR/bin/kafka-broker-api-versions.sh \ - --bootstrap-server localhost:9092 > /dev/null 2>&1 \ - && echo "Broker ready after ${!i}x5s" && break \ - || sleep 5 - done - - # Enable KIP-932 share groups - $KAFKA_DIR/bin/kafka-features.sh \ - --bootstrap-server localhost:9092 \ - upgrade --feature share.version=1 - echo "Share groups (KIP-932) enabled." - - echo "=== Setup complete $(date). Bootstrap: $INSTANCE_IP:9092 ===" - signal_cfn 0 - - CreationPolicy: - ResourceSignal: - Timeout: PT40M - - # Dummy resource to ensure IGW is attached before EC2 launches - VPCGatewayAttachmentDependency: - Type: AWS::CloudFormation::WaitConditionHandle Outputs: BootstrapServers: From df5d5a2564546a3132c35d66ebb32df8be6deba8 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 15:21:12 -0400 Subject: [PATCH 16/28] fix: Increase wget timeout in setup-broker.sh to 40 min --- smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh b/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh index 9ca41177b..c15332189 100755 --- a/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh +++ b/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh @@ -113,7 +113,7 @@ run_ssm "1/6 Install Java" \ # ── Step 2: Download Kafka ──────────────────────────────────── run_ssm "2/6 Download Kafka $KAFKA_VERSION (this takes ~20 min)" \ "wget -q --timeout=1800 --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" \ - 1800 + 2400 # ── Step 3: Extract ─────────────────────────────────────────── run_ssm "3/6 Extract Kafka" \ From 7ba08db5eb4f5affcbbef48fb6fc93cf9b93e3d0 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 15:21:55 -0400 Subject: [PATCH 17/28] fix: Align wget --timeout with SSM poll timeout (both 40 min) --- smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh b/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh index c15332189..0edfeb95e 100755 --- a/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh +++ b/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh @@ -112,7 +112,7 @@ run_ssm "1/6 Install Java" \ # ── Step 2: Download Kafka ──────────────────────────────────── run_ssm "2/6 Download Kafka $KAFKA_VERSION (this takes ~20 min)" \ - "wget -q --timeout=1800 --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" \ + "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 ─────────────────────────────────────────── From b6916f235eab3ce445bcd96dc9387a5f14ccdb64 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 16:00:21 -0400 Subject: [PATCH 18/28] fix: Restore missing Path B heading in README --- smk-lambda-queue-mode-python-sam/README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index c878c6b4d..fbaa6cd5e 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -129,6 +129,10 @@ aws cloudformation deploy \ 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 From 966d270f5944bfa4f62a59ca5ac31023576e1b20 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 16:03:33 -0400 Subject: [PATCH 19/28] fix: Remove double quotes from advertised.listeners in setup-broker.sh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Double quotes inside the SSM --parameters JSON string caused a ParamValidation error. Removed surrounding quotes from the advertised.listeners value — no quotes needed since there are no spaces. Validated end-to-end: pattern deploys successfully and processes records with Queue mode on a fresh account deployment. --- smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh b/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh index 0edfeb95e..7ef6d3bba 100755 --- a/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh +++ b/smk-lambda-queue-mode-python-sam/scripts/setup-broker.sh @@ -122,7 +122,7 @@ run_ssm "3/6 Extract Kafka" \ # ── 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" \ + "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 ──────────────────── From c9b9d2d9a7c28651d3185728803e3d3f66cadbca Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 17:37:12 -0400 Subject: [PATCH 20/28] =?UTF-8?q?fix:=20Simplify=20VPC=20endpoint=20cleanu?= =?UTF-8?q?p=20=E2=80=94=20derive=20VPC=20ID=20from=20stack=20output?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- smk-lambda-queue-mode-python-sam/README.md | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index fbaa6cd5e..88708e079 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -327,12 +327,16 @@ Delete stacks in reverse order: aws lambda delete-event-source-mapping --uuid --region # 2. Delete VPC endpoints -aws ec2 describe-vpc-endpoints \ - --filters "Name=vpc-id,Values=" \ +VPC_ID=$(aws cloudformation describe-stacks \ + --stack-name kafka-queue-network \ + --query 'Stacks[0].Outputs[?OutputKey==`VpcId`].OutputValue' \ + --output text --region ) +ENDPOINT_IDS=$(aws ec2 describe-vpc-endpoints \ + --filters "Name=vpc-id,Values=$VPC_ID" \ "Name=service-name,Values=com.amazonaws..lambda,com.amazonaws..sts,com.amazonaws..sqs" \ - --query 'VpcEndpoints[*].VpcEndpointId' --output text --region -# Then delete each endpoint: -aws ec2 delete-vpc-endpoints --vpc-endpoint-ids --region + --query 'VpcEndpoints[*].VpcEndpointId' --output text --region ) +aws ec2 delete-vpc-endpoints \ + --vpc-endpoint-ids $ENDPOINT_IDS --region # 3. Delete application stacks aws cloudformation delete-stack --stack-name kafka-queue-observability --region From bd5bdef5d82ba1e70c404c8d6901135732a61162 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 17:38:57 -0400 Subject: [PATCH 21/28] fix: Use REGION variable in VPC endpoint cleanup command --- smk-lambda-queue-mode-python-sam/README.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index 88708e079..79a9312f5 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -327,16 +327,17 @@ Delete stacks in reverse order: aws lambda delete-event-source-mapping --uuid --region # 2. Delete VPC endpoints +REGION= VPC_ID=$(aws cloudformation describe-stacks \ --stack-name kafka-queue-network \ --query 'Stacks[0].Outputs[?OutputKey==`VpcId`].OutputValue' \ - --output text --region ) + --output text --region $REGION) ENDPOINT_IDS=$(aws ec2 describe-vpc-endpoints \ --filters "Name=vpc-id,Values=$VPC_ID" \ - "Name=service-name,Values=com.amazonaws..lambda,com.amazonaws..sts,com.amazonaws..sqs" \ - --query 'VpcEndpoints[*].VpcEndpointId' --output text --region ) + "Name=service-name,Values=com.amazonaws.$REGION.lambda,com.amazonaws.$REGION.sts,com.amazonaws.$REGION.sqs" \ + --query 'VpcEndpoints[*].VpcEndpointId' --output text --region $REGION) aws ec2 delete-vpc-endpoints \ - --vpc-endpoint-ids $ENDPOINT_IDS --region + --vpc-endpoint-ids $ENDPOINT_IDS --region $REGION # 3. Delete application stacks aws cloudformation delete-stack --stack-name kafka-queue-observability --region From 984ae14ce8e4b8ea210a43090a2c1d7ed86ca093 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 17:40:20 -0400 Subject: [PATCH 22/28] fix: Revert to placeholder in cleanup, add inline note --- smk-lambda-queue-mode-python-sam/README.md | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index 79a9312f5..297145844 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -326,18 +326,17 @@ Delete stacks in reverse order: # 1. Delete the ESM first (get UUID from create-esm.sh output or console) aws lambda delete-event-source-mapping --uuid --region -# 2. Delete VPC endpoints -REGION= +# 2. Delete VPC endpoints (replace with your region, e.g. us-east-1) VPC_ID=$(aws cloudformation describe-stacks \ --stack-name kafka-queue-network \ --query 'Stacks[0].Outputs[?OutputKey==`VpcId`].OutputValue' \ - --output text --region $REGION) + --output text --region ) ENDPOINT_IDS=$(aws ec2 describe-vpc-endpoints \ --filters "Name=vpc-id,Values=$VPC_ID" \ - "Name=service-name,Values=com.amazonaws.$REGION.lambda,com.amazonaws.$REGION.sts,com.amazonaws.$REGION.sqs" \ - --query 'VpcEndpoints[*].VpcEndpointId' --output text --region $REGION) + "Name=service-name,Values=com.amazonaws..lambda,com.amazonaws..sts,com.amazonaws..sqs" \ + --query 'VpcEndpoints[*].VpcEndpointId' --output text --region ) aws ec2 delete-vpc-endpoints \ - --vpc-endpoint-ids $ENDPOINT_IDS --region $REGION + --vpc-endpoint-ids $ENDPOINT_IDS --region # 3. Delete application stacks aws cloudformation delete-stack --stack-name kafka-queue-observability --region From dc07d844eb0486a669b0bc98ad63cf3b15cf107d Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 17:42:34 -0400 Subject: [PATCH 23/28] fix: Simplify VPC endpoint cleanup to console instructions --- smk-lambda-queue-mode-python-sam/README.md | 14 +++----------- 1 file changed, 3 insertions(+), 11 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index 297145844..900f73b8a 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -326,17 +326,9 @@ Delete stacks in reverse order: # 1. Delete the ESM first (get UUID from create-esm.sh output or console) aws lambda delete-event-source-mapping --uuid --region -# 2. Delete VPC endpoints (replace with your region, e.g. us-east-1) -VPC_ID=$(aws cloudformation describe-stacks \ - --stack-name kafka-queue-network \ - --query 'Stacks[0].Outputs[?OutputKey==`VpcId`].OutputValue' \ - --output text --region ) -ENDPOINT_IDS=$(aws ec2 describe-vpc-endpoints \ - --filters "Name=vpc-id,Values=$VPC_ID" \ - "Name=service-name,Values=com.amazonaws..lambda,com.amazonaws..sts,com.amazonaws..sqs" \ - --query 'VpcEndpoints[*].VpcEndpointId' --output text --region ) -aws ec2 delete-vpc-endpoints \ - --vpc-endpoint-ids $ENDPOINT_IDS --region +# 2. Delete VPC endpoints +# Open the VPC console → Endpoints, filter by your VPC ID, +# select the lambda, sts, and sqs endpoints, then Actions → Delete. # 3. Delete application stacks aws cloudformation delete-stack --stack-name kafka-queue-observability --region From 340f007286ca536b1fb71fe1ae80481735bbe210 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 17:44:12 -0400 Subject: [PATCH 24/28] fix: Break VPC endpoint cleanup into simple numbered steps --- smk-lambda-queue-mode-python-sam/README.md | 25 ++++++++++++++++------ 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index 900f73b8a..fbeef4e02 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -326,18 +326,31 @@ Delete stacks in reverse order: # 1. Delete the ESM first (get UUID from create-esm.sh output or console) aws lambda delete-event-source-mapping --uuid --region -# 2. Delete VPC endpoints -# Open the VPC console → Endpoints, filter by your VPC ID, -# select the lambda, sts, and sqs endpoints, then Actions → Delete. +# 2. Get your VPC ID +aws cloudformation describe-stacks \ + --stack-name kafka-queue-network \ + --query 'Stacks[0].Outputs[?OutputKey==`VpcId`].OutputValue' \ + --output text --region + +# 3. List VPC endpoint IDs (replace with output from step 2) +aws ec2 describe-vpc-endpoints \ + --filters "Name=vpc-id,Values=" \ + --query 'VpcEndpoints[?contains(ServiceName,`lambda`) || contains(ServiceName,`sts`) || contains(ServiceName,`sqs`)].[VpcEndpointId,ServiceName]' \ + --output text --region + +# 4. Delete endpoints (replace with space-separated IDs from step 3) +aws ec2 delete-vpc-endpoints \ + --vpc-endpoint-ids \ + --region -# 3. Delete application stacks +# 5. Delete application stacks aws cloudformation delete-stack --stack-name kafka-queue-observability --region aws cloudformation delete-stack --stack-name kafka-queue-app --region -# 4. Delete broker (if deployed) +# 6. Delete broker (if deployed) aws cloudformation delete-stack --stack-name kafka-queue-broker --region -# 5. Delete network (if deployed) +# 7. Delete network (if deployed) aws cloudformation delete-stack --stack-name kafka-queue-network --region ``` From 3216f429b34522eda79df17c20e51abf21f0c9fe Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 17:46:13 -0400 Subject: [PATCH 25/28] fix: Break VPC endpoint cleanup into one service at a time --- smk-lambda-queue-mode-python-sam/README.md | 25 +++++++++++----------- 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index fbeef4e02..2648909a8 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -332,25 +332,26 @@ aws cloudformation describe-stacks \ --query 'Stacks[0].Outputs[?OutputKey==`VpcId`].OutputValue' \ --output text --region -# 3. List VPC endpoint IDs (replace with output from step 2) -aws ec2 describe-vpc-endpoints \ - --filters "Name=vpc-id,Values=" \ - --query 'VpcEndpoints[?contains(ServiceName,`lambda`) || contains(ServiceName,`sts`) || contains(ServiceName,`sqs`)].[VpcEndpointId,ServiceName]' \ - --output text --region +# 3. Find and delete the lambda VPC endpoint +aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=" "Name=service-name,Values=com.amazonaws..lambda" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region +aws ec2 delete-vpc-endpoints --vpc-endpoint-ids --region -# 4. Delete endpoints (replace with space-separated IDs from step 3) -aws ec2 delete-vpc-endpoints \ - --vpc-endpoint-ids \ - --region +# 4. Find and delete the sts VPC endpoint +aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=" "Name=service-name,Values=com.amazonaws..sts" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region +aws ec2 delete-vpc-endpoints --vpc-endpoint-ids --region + +# 5. Find and delete the sqs VPC endpoint +aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=" "Name=service-name,Values=com.amazonaws..sqs" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region +aws ec2 delete-vpc-endpoints --vpc-endpoint-ids --region -# 5. Delete application stacks +# 6. Delete application stacks aws cloudformation delete-stack --stack-name kafka-queue-observability --region aws cloudformation delete-stack --stack-name kafka-queue-app --region -# 6. Delete broker (if deployed) +# 7. Delete broker (if deployed) aws cloudformation delete-stack --stack-name kafka-queue-broker --region -# 7. Delete network (if deployed) +# 8. Delete network (if deployed) aws cloudformation delete-stack --stack-name kafka-queue-network --region ``` From 036219f88d18528c1b81ad9ba24287f6bb3a69ec Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 18:13:34 -0400 Subject: [PATCH 26/28] fix: Correct cleanup instructions in example-pattern.json --- smk-lambda-queue-mode-python-sam/example-pattern.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smk-lambda-queue-mode-python-sam/example-pattern.json b/smk-lambda-queue-mode-python-sam/example-pattern.json index 156ee4018..7982a3f8a 100644 --- a/smk-lambda-queue-mode-python-sam/example-pattern.json +++ b/smk-lambda-queue-mode-python-sam/example-pattern.json @@ -56,7 +56,7 @@ }, "cleanup": { "text": [ - "Delete stacks in reverse order: aws cloudformation delete-stack --stack-name kafka-queue-esm, then kafka-queue-app, kafka-queue-broker, kafka-queue-network." + "Delete the ESM first: aws lambda delete-event-source-mapping --uuid <esm-uuid>, then delete stacks in reverse order: kafka-queue-observability, kafka-queue-app, kafka-queue-broker, kafka-queue-network." ] }, "authors": [ From c8e8594717ed55206d92a63f2f7e365cb71a9358 Mon Sep 17 00:00:00 2001 From: vaibhav-jain-lilly Date: Thu, 17 Sep 2026 18:19:11 -0400 Subject: [PATCH 27/28] Clean up --- smk-lambda-queue-mode-python-sam/README.md | 4 ++++ smk-lambda-queue-mode-python-sam/example-pattern.json | 8 ++------ 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index 2648909a8..a3bac2165 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -334,18 +334,22 @@ aws cloudformation describe-stacks \ # 3. Find and delete the lambda VPC endpoint aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=" "Name=service-name,Values=com.amazonaws..lambda" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region + aws ec2 delete-vpc-endpoints --vpc-endpoint-ids --region # 4. Find and delete the sts VPC endpoint aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=" "Name=service-name,Values=com.amazonaws..sts" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region + aws ec2 delete-vpc-endpoints --vpc-endpoint-ids --region # 5. Find and delete the sqs VPC endpoint aws ec2 describe-vpc-endpoints --filters "Name=vpc-id,Values=" "Name=service-name,Values=com.amazonaws..sqs" --query 'VpcEndpoints[0].VpcEndpointId' --output text --region + aws ec2 delete-vpc-endpoints --vpc-endpoint-ids --region # 6. Delete application stacks aws cloudformation delete-stack --stack-name kafka-queue-observability --region + aws cloudformation delete-stack --stack-name kafka-queue-app --region # 7. Delete broker (if deployed) diff --git a/smk-lambda-queue-mode-python-sam/example-pattern.json b/smk-lambda-queue-mode-python-sam/example-pattern.json index 7982a3f8a..6bea56ba3 100644 --- a/smk-lambda-queue-mode-python-sam/example-pattern.json +++ b/smk-lambda-queue-mode-python-sam/example-pattern.json @@ -31,15 +31,11 @@ "link": "https://docs.aws.amazon.com/lambda/latest/dg/with-kafka.html" }, { - "text": "Apache Kafka KIP-932: Queues for Kafka", - "link": "https://cwiki.apache.org/confluence/display/KAFKA/KIP-932+Queues+for+Kafka" - }, - { - "text": "Lambda ESM provisioned mode for Kafka", + "text": "Apache Kafka event poller scaling modes in Lambda", "link": "https://docs.aws.amazon.com/lambda/latest/dg/kafka-scaling-modes.html" }, { - "text": "Partial batch response for Lambda", + "text": "Configuring error handling controls for Kafka event sources", "link": "https://docs.aws.amazon.com/lambda/latest/dg/kafka-retry-configurations.html" } ] From c75656eac848ae21ce1d7f1f816845cbcfe1d180 Mon Sep 17 00:00:00 2001 From: Hardith Murari Date: Thu, 17 Sep 2026 16:42:41 -0700 Subject: [PATCH 28/28] docs: add Hardith Murari as co-author and drop README Author section --- smk-lambda-queue-mode-python-sam/README.md | 6 ------ smk-lambda-queue-mode-python-sam/example-pattern.json | 5 +++++ 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/smk-lambda-queue-mode-python-sam/README.md b/smk-lambda-queue-mode-python-sam/README.md index a3bac2165..529393152 100644 --- a/smk-lambda-queue-mode-python-sam/README.md +++ b/smk-lambda-queue-mode-python-sam/README.md @@ -372,9 +372,3 @@ aws cloudformation delete-stack --stack-name kafka-queue-network --region