From 7ba814c2efba739f32491c430671e68bcd234d47 Mon Sep 17 00:00:00 2001 From: Nithin Chandran Rajashankar Date: Mon, 13 Jul 2026 06:06:45 +0000 Subject: [PATCH 1/2] feat(dsql-cdc-eventbridge-fanout-cdk): DSQL CDC to EventBridge fan-out Event-driven microservices pattern using Amazon Aurora DSQL CDC (GA July 2026) as a change event source. CDC events flow through Amazon Kinesis to AWS Lambda, which classifies operations and publishes typed events to Amazon EventBridge. Three independent consumers demonstrate content-based routing: Amazon SQS (audit), AWS Step Functions (validation), Amazon SNS (alerting). 7 AWS services, 29 CFN resources. Fundamentally different from existing apigw-lambda-dsql (basic CRUD) and sample-aurora-dsql-cdc-demo (RAG/semantic search). This pattern shows DSQL as an event source for microservices fan-out via EventBridge. --- dsql-cdc-eventbridge-fanout-cdk/README.md | 148 ++++++++++ .../cdk/.gitignore | 6 + .../cdk/bin/app.ts | 12 + dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json | 3 + .../lib/dsql-cdc-eventbridge-fanout-stack.ts | 278 ++++++++++++++++++ .../cdk/package.json | 20 ++ .../cdk/tsconfig.json | 24 ++ .../example-pattern.json | 113 +++++++ .../lambdas/cdc-processor/handler.py | 78 +++++ .../lambdas/cdc-stream-manager/handler.py | 108 +++++++ 10 files changed, 790 insertions(+) create mode 100644 dsql-cdc-eventbridge-fanout-cdk/README.md create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/package.json create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json create mode 100644 dsql-cdc-eventbridge-fanout-cdk/example-pattern.json create mode 100644 dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py create mode 100644 dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py diff --git a/dsql-cdc-eventbridge-fanout-cdk/README.md b/dsql-cdc-eventbridge-fanout-cdk/README.md new file mode 100644 index 0000000000..0c03a529ed --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/README.md @@ -0,0 +1,148 @@ +# Amazon Aurora DSQL CDC to Amazon EventBridge Fan-out + +This pattern deploys an event-driven fan-out architecture that captures real-time database changes from Amazon Aurora DSQL using Change Data Capture (CDC), streams them through Amazon Kinesis Data Streams, processes them with AWS Lambda, and routes them to multiple consumers via Amazon EventBridge. + +Important: This pattern is fundamentally different from the existing `apigw-lambda-dsql` pattern (basic CRUD) and the standalone [sample-aurora-dsql-cdc-demo](https://github.com/aws-samples/sample-aurora-dsql-cdc-demo) (RAG/semantic search chatbot). This pattern demonstrates **event-driven microservices fan-out** — using Amazon Aurora DSQL CDC as a change event source to drive multiple independent downstream workflows through Amazon EventBridge content-based routing. + +Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/dsql-cdc-eventbridge-fanout-cdk + +## Architecture + +``` +┌──────────────┐ ┌─────────────────────┐ ┌──────────────────────┐ ┌─────────────────────┐ +│ Amazon Aurora │────▶│ Amazon Kinesis Data │────▶│ AWS Lambda │────▶│ Amazon EventBridge │ +│ DSQL (CDC) │ │ Streams │ │ (CDC Processor) │ │ (Custom Bus) │ +└──────────────┘ └─────────────────────┘ └──────────────────────┘ └──────────┬──────────┘ + │ + ┌────────────────────────────────────────────────────────────┼───────────────────┐ + │ │ │ + ▼ ▼ ▼ + ┌──────────────────┐ ┌─────────────────┐ ┌──────────────┐ + │ Amazon SQS │ │ AWS Step │ │ Amazon SNS │ + │ (Audit Queue) │ │ Functions │ │ (Alerts) │ + │ ALL events │ │ INSERT events │ │ DELETE events │ + └──────────────────┘ └─────────────────┘ └──────────────┘ +``` + +**How it works:** + +1. Amazon Aurora DSQL captures every committed row-level change (INSERT, UPDATE, DELETE) and delivers it as a structured JSON record to Amazon Kinesis Data Streams +2. AWS Lambda consumes the Amazon Kinesis stream, parses the CDC payload, classifies the operation type, and publishes typed events to an Amazon EventBridge custom event bus +3. Amazon EventBridge routes events to three independent consumers based on content: + - **ALL changes** → Amazon SQS audit queue (compliance/audit trail) + - **INSERT events** → AWS Step Functions workflow (data validation) + - **DELETE events** → Amazon SNS topic (real-time alerting) + +## Requirements + +- [AWS CDK v2](https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html) installed and configured +- [Node.js 20+](https://nodejs.org/) with npm +- AWS account [bootstrapped for CDK](https://docs.aws.amazon.com/cdk/v2/guide/bootstrapping.html) +- An existing Amazon Aurora DSQL cluster (see Deployment section) +- Python 3.12 (for AWS Lambda functions) + +## Deployment + +### Step 1: Create an Amazon Aurora DSQL cluster (if you don't have one) + +```bash +aws dsql create-cluster --region us-east-1 +``` + +Note the `identifier` from the response — you'll need it in Step 3. + +### Step 2: Install dependencies and synthesize + +```bash +cd dsql-cdc-eventbridge-fanout-cdk/cdk +npm install +npx cdk synth +``` + +### Step 3: Deploy the stack + +```bash +npx cdk deploy --parameters DsqlClusterId= +``` + +Replace `` with your Amazon Aurora DSQL cluster identifier. + +### Step 4: Insert data to trigger CDC events + +Connect to your Amazon Aurora DSQL cluster and insert data: + +```sql +CREATE TABLE orders ( + id UUID DEFAULT gen_random_uuid() PRIMARY KEY, + customer_name TEXT NOT NULL, + amount DECIMAL(10,2), + created_at TIMESTAMP DEFAULT now() +); + +INSERT INTO orders (customer_name, amount) VALUES ('Acme Corp', 1250.00); +INSERT INTO orders (customer_name, amount) VALUES ('Globex Inc', 780.50); +DELETE FROM orders WHERE customer_name = 'Globex Inc'; +``` + +## Testing + +After deploying, verify the fan-out is working: + +### 1. Check Amazon SQS audit queue (receives ALL events) + +```bash +aws sqs receive-message \ + --queue-url $(aws cloudformation describe-stacks \ + --stack-name DsqlCdcEventbridgeFanoutStack \ + --query 'Stacks[0].Outputs[?OutputKey==`AuditQueueUrl`].OutputValue' \ + --output text) \ + --max-number-of-messages 5 +``` + +### 2. Check AWS Step Functions executions (INSERT events only) + +```bash +aws stepfunctions list-executions \ + --state-machine-arn $(aws cloudformation describe-stacks \ + --stack-name DsqlCdcEventbridgeFanoutStack \ + --query 'Stacks[0].Outputs[?OutputKey==`ValidationStateMachineArn`].OutputValue' \ + --output text) \ + --max-results 5 +``` + +### 3. Subscribe to Amazon SNS topic to receive DELETE alerts + +```bash +aws sns subscribe \ + --topic-arn $(aws cloudformation describe-stacks \ + --stack-name DsqlCdcEventbridgeFanoutStack \ + --query 'Stacks[0].Outputs[?OutputKey==`AlertTopicArn`].OutputValue' \ + --output text) \ + --protocol email \ + --notification-endpoint your-email@example.com +``` + +## Cleanup + +> **Warning:** This will delete all resources including the Amazon SQS queues and their messages. The Amazon Aurora DSQL cluster is NOT deleted (it was created externally). + +```bash +cd dsql-cdc-eventbridge-fanout-cdk/cdk +npx cdk destroy +``` + +## Services Used + +| Service | Role | +|---------|------| +| Amazon Aurora DSQL | Source database with CDC enabled | +| Amazon Kinesis Data Streams | Receives CDC event stream from Amazon Aurora DSQL | +| AWS Lambda | Processes CDC events, classifies operations, publishes to Amazon EventBridge | +| Amazon EventBridge | Content-based routing of CDC events to multiple consumers | +| Amazon SQS | Audit queue for all change events (compliance) | +| AWS Step Functions | Data validation workflow for INSERT events | +| Amazon SNS | Real-time alerting for DELETE events | + +---- +Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +SPDX-License-Identifier: MIT-0 diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore b/dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore new file mode 100644 index 0000000000..1f0ea03523 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore @@ -0,0 +1,6 @@ +node_modules +cdk.out +cdk.context.json +build +*.js +*.d.ts diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts b/dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts new file mode 100644 index 0000000000..ea4b329a03 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts @@ -0,0 +1,12 @@ +#!/usr/bin/env node +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT-0 (2026) + +import 'source-map-support/register'; +import * as cdk from 'aws-cdk-lib'; +import { DsqlCdcEventbridgeFanoutStack } from '../lib/dsql-cdc-eventbridge-fanout-stack'; + +const app = new cdk.App(); +new DsqlCdcEventbridgeFanoutStack(app, 'DsqlCdcEventbridgeFanoutStack', { + description: 'Amazon Aurora DSQL CDC to Amazon EventBridge fan-out pattern (uksb-1tupboc57)', +}); diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json b/dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json new file mode 100644 index 0000000000..a6700a2ff4 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json @@ -0,0 +1,3 @@ +{ + "app": "npx ts-node --prefer-ts-exts bin/app.ts" +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts b/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts new file mode 100644 index 0000000000..d12cc6ee30 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts @@ -0,0 +1,278 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT-0 (2026) + +import * as cdk from 'aws-cdk-lib'; +import * as kinesis from 'aws-cdk-lib/aws-kinesis'; +import * as lambda from 'aws-cdk-lib/aws-lambda'; +import * as iam from 'aws-cdk-lib/aws-iam'; +import * as events from 'aws-cdk-lib/aws-events'; +import * as targets from 'aws-cdk-lib/aws-events-targets'; +import * as sqs from 'aws-cdk-lib/aws-sqs'; +import * as sns from 'aws-cdk-lib/aws-sns'; +import * as snsSubs from 'aws-cdk-lib/aws-sns-subscriptions'; +import * as sfn from 'aws-cdk-lib/aws-stepfunctions'; +import * as sfnTasks from 'aws-cdk-lib/aws-stepfunctions-tasks'; +import * as logs from 'aws-cdk-lib/aws-logs'; +import * as cr from 'aws-cdk-lib/custom-resources'; +import * as kinesisEvtSrc from 'aws-cdk-lib/aws-lambda-event-sources'; +import { Construct } from 'constructs'; +import * as path from 'path'; + +export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { + constructor(scope: Construct, id: string, props?: cdk.StackProps) { + super(scope, id, props); + + const region = cdk.Stack.of(this).region; + const account = cdk.Stack.of(this).account; + + // --- Context parameters --- + const dsqlClusterId = new cdk.CfnParameter(this, 'DsqlClusterId', { + type: 'String', + description: 'Amazon Aurora DSQL cluster identifier', + }); + + // ========================================================= + // 1. Amazon Kinesis Data Stream (CDC target) + // ========================================================= + const cdcStream = new kinesis.Stream(this, 'CdcStream', { + streamName: `dsql-cdc-${cdk.Names.uniqueId(this).slice(-8).toLowerCase()}`, + shardCount: 1, + retentionPeriod: cdk.Duration.hours(24), + encryption: kinesis.StreamEncryption.MANAGED, + }); + + // ========================================================= + // 2. IAM Role for DSQL to write to Kinesis + // ========================================================= + const dsqlCdcRole = new iam.Role(this, 'DsqlCdcRole', { + assumedBy: new iam.ServicePrincipal('dsql.amazonaws.com'), + description: 'Allows Amazon Aurora DSQL CDC to write to Amazon Kinesis', + inlinePolicies: { + KinesisWrite: new iam.PolicyDocument({ + statements: [ + new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + actions: [ + 'kinesis:PutRecord', + 'kinesis:PutRecords', + 'kinesis:DescribeStream', + ], + resources: [cdcStream.streamArn], + }), + ], + }), + }, + }); + + // ========================================================= + // 3. Custom Resource: Create/Delete DSQL CDC Stream + // (No CFN resource type for DSQL streams yet — use SDK) + // ========================================================= + const streamManagerFn = new lambda.Function(this, 'CdcStreamManagerFn', { + runtime: lambda.Runtime.PYTHON_3_12, + handler: 'handler.on_event', + code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/cdc-stream-manager')), + timeout: cdk.Duration.minutes(5), + memorySize: 256, + description: 'Custom Resource: manages Amazon Aurora DSQL CDC stream lifecycle', + environment: { + CLUSTER_ID: dsqlClusterId.valueAsString, + KINESIS_STREAM_ARN: cdcStream.streamArn, + CDC_ROLE_ARN: dsqlCdcRole.roleArn, + }, + }); + + streamManagerFn.addToRolePolicy(new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + actions: [ + 'dsql:CreateStream', + 'dsql:DeleteStream', + 'dsql:GetStream', + 'dsql:ListStreams', + ], + resources: [ + `arn:aws:dsql:${region}:${account}:cluster/${dsqlClusterId.valueAsString}`, + `arn:aws:dsql:${region}:${account}:cluster/${dsqlClusterId.valueAsString}/stream/*`, + ], + })); + + streamManagerFn.addToRolePolicy(new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + actions: ['iam:PassRole'], + resources: [dsqlCdcRole.roleArn], + conditions: { + StringEquals: { 'iam:PassedToService': 'dsql.amazonaws.com' }, + }, + })); + + const cdcStreamCr = new cr.Provider(this, 'CdcStreamProvider', { + onEventHandler: streamManagerFn, + }); + + const cdcStreamResource = new cdk.CustomResource(this, 'DsqlCdcStream', { + serviceToken: cdcStreamCr.serviceToken, + properties: { + ClusterId: dsqlClusterId.valueAsString, + KinesisStreamArn: cdcStream.streamArn, + RoleArn: dsqlCdcRole.roleArn, + }, + }); + + // ========================================================= + // 4. Amazon EventBridge Custom Event Bus + // ========================================================= + const cdcEventBus = new events.EventBus(this, 'CdcEventBus', { + eventBusName: 'dsql-cdc-events', + }); + + // ========================================================= + // 5. AWS Lambda: CDC Processor (Kinesis → EventBridge) + // ========================================================= + const cdcProcessorFn = new lambda.Function(this, 'CdcProcessorFn', { + runtime: lambda.Runtime.PYTHON_3_12, + handler: 'handler.lambda_handler', + code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/cdc-processor')), + timeout: cdk.Duration.seconds(60), + memorySize: 256, + description: 'Processes Amazon Aurora DSQL CDC events and fans out to Amazon EventBridge', + environment: { + EVENT_BUS_NAME: cdcEventBus.eventBusName, + }, + }); + + cdcProcessorFn.addToRolePolicy(new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + actions: ['events:PutEvents'], + resources: [cdcEventBus.eventBusArn], + })); + + cdcProcessorFn.addEventSource(new kinesisEvtSrc.KinesisEventSource(cdcStream, { + startingPosition: lambda.StartingPosition.TRIM_HORIZON, + batchSize: 100, + maxBatchingWindow: cdk.Duration.seconds(5), + retryAttempts: 3, + bisectBatchOnError: true, + })); + + // ========================================================= + // 6. Fan-out Target A: Amazon SQS (Audit Queue) + // ========================================================= + const auditDlq = new sqs.Queue(this, 'AuditDlq', { + queueName: 'dsql-cdc-audit-dlq', + retentionPeriod: cdk.Duration.days(14), + encryption: sqs.QueueEncryption.SQS_MANAGED, + }); + + const auditQueue = new sqs.Queue(this, 'AuditQueue', { + queueName: 'dsql-cdc-audit', + visibilityTimeout: cdk.Duration.seconds(30), + encryption: sqs.QueueEncryption.SQS_MANAGED, + deadLetterQueue: { queue: auditDlq, maxReceiveCount: 3 }, + }); + + new events.Rule(this, 'AuditRule', { + eventBus: cdcEventBus, + ruleName: 'route-all-changes-to-audit', + description: 'Route ALL CDC events to Amazon SQS audit queue', + eventPattern: { + source: ['dsql.cdc'], + }, + targets: [new targets.SqsQueue(auditQueue)], + }); + + // ========================================================= + // 7. Fan-out Target B: AWS Step Functions (Data Validation) + // ========================================================= + const validatePass = new sfn.Pass(this, 'ValidationPassed', { + result: sfn.Result.fromObject({ validated: true }), + }); + + const validateFail = new sfn.Pass(this, 'ValidationFailed', { + result: sfn.Result.fromObject({ validated: false }), + }); + + const validationChoice = new sfn.Choice(this, 'HasRequiredFields') + .when( + sfn.Condition.isPresent('$.detail.newImage'), + validatePass, + ) + .otherwise(validateFail); + + const validationStateMachine = new sfn.StateMachine(this, 'ValidationWorkflow', { + definitionBody: sfn.DefinitionBody.fromChainable(validationChoice), + timeout: cdk.Duration.minutes(5), + tracingEnabled: true, + logs: { + destination: new logs.LogGroup(this, 'ValidationLogGroup', { + logGroupName: '/aws/stepfunctions/dsql-cdc-validation', + retention: logs.RetentionDays.ONE_WEEK, + removalPolicy: cdk.RemovalPolicy.DESTROY, + }), + level: sfn.LogLevel.ERROR, + }, + }); + + new events.Rule(this, 'ValidationRule', { + eventBus: cdcEventBus, + ruleName: 'route-inserts-to-validation', + description: 'Route INSERT events to AWS Step Functions for validation', + eventPattern: { + source: ['dsql.cdc'], + detailType: ['INSERT'], + }, + targets: [new targets.SfnStateMachine(validationStateMachine)], + }); + + // ========================================================= + // 8. Fan-out Target C: Amazon SNS (Real-time Alerts) + // ========================================================= + const alertTopic = new sns.Topic(this, 'AlertTopic', { + topicName: 'dsql-cdc-delete-alerts', + displayName: 'Amazon Aurora DSQL CDC - DELETE Alerts', + }); + + new events.Rule(this, 'DeleteAlertRule', { + eventBus: cdcEventBus, + ruleName: 'route-deletes-to-alert', + description: 'Route DELETE events to Amazon SNS for alerting', + eventPattern: { + source: ['dsql.cdc'], + detailType: ['DELETE'], + }, + targets: [new targets.SnsTopic(alertTopic)], + }); + + // ========================================================= + // Outputs + // ========================================================= + new cdk.CfnOutput(this, 'KinesisStreamArn', { + value: cdcStream.streamArn, + description: 'Amazon Kinesis Data Stream ARN (CDC target)', + }); + + new cdk.CfnOutput(this, 'EventBusArn', { + value: cdcEventBus.eventBusArn, + description: 'Amazon EventBridge custom event bus ARN', + }); + + new cdk.CfnOutput(this, 'AuditQueueUrl', { + value: auditQueue.queueUrl, + description: 'Amazon SQS audit queue URL', + }); + + new cdk.CfnOutput(this, 'ValidationStateMachineArn', { + value: validationStateMachine.stateMachineArn, + description: 'AWS Step Functions validation workflow ARN', + }); + + new cdk.CfnOutput(this, 'AlertTopicArn', { + value: alertTopic.topicArn, + description: 'Amazon SNS alert topic ARN', + }); + + new cdk.CfnOutput(this, 'CdcStreamId', { + value: cdcStreamResource.getAttString('StreamId'), + description: 'Amazon Aurora DSQL CDC stream identifier', + }); + } +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/package.json b/dsql-cdc-eventbridge-fanout-cdk/cdk/package.json new file mode 100644 index 0000000000..6af041f45b --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/package.json @@ -0,0 +1,20 @@ +{ + "name": "dsql-cdc-eventbridge-fanout-cdk", + "version": "1.0.0", + "bin": { "app": "bin/app.ts" }, + "scripts": { + "build": "tsc", + "synth": "cdk synth", + "deploy": "cdk deploy", + "destroy": "cdk destroy" + }, + "dependencies": { + "aws-cdk-lib": "2.185.0", + "constructs": "^10.0.0", + "source-map-support": "^0.5.21" + }, + "devDependencies": { + "typescript": "~5.4.0", + "ts-node": "^10.9.0" + } +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json b/dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json new file mode 100644 index 0000000000..7ddcfe705e --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json @@ -0,0 +1,24 @@ +{ + "compilerOptions": { + "target": "ES2020", + "module": "commonjs", + "lib": ["es2020"], + "declaration": true, + "strict": true, + "noImplicitAny": true, + "strictNullChecks": true, + "noImplicitThis": true, + "alwaysStrict": true, + "noUnusedLocals": false, + "noUnusedParameters": false, + "noImplicitReturns": true, + "noFallthroughCasesInSwitch": false, + "inlineSourceMap": true, + "inlineSources": true, + "experimentalDecorators": true, + "strictPropertyInitialization": false, + "outDir": "./build", + "rootDir": "." + }, + "exclude": ["node_modules", "build"] +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json b/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json new file mode 100644 index 0000000000..4e7fbc86c0 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json @@ -0,0 +1,113 @@ +{ + "title": "Amazon Aurora DSQL CDC to Amazon EventBridge Fan-out", + "description": "Stream real-time database changes from Amazon Aurora DSQL via CDC to Amazon EventBridge for content-based routing to multiple consumers", + "language": "TypeScript", + "level": "300", + "framework": "AWS CDK", + "introBox": { + "headline": "How it works", + "text": [ + "Amazon Aurora DSQL captures committed row-level changes (INSERT, UPDATE, DELETE) and streams them as JSON records to Amazon Kinesis Data Streams.", + "AWS Lambda consumes the stream, classifies each CDC event by operation type, and publishes typed events to an Amazon EventBridge custom event bus.", + "Amazon EventBridge routes events to three independent consumers: Amazon SQS (audit trail for ALL events), AWS Step Functions (validation workflow for INSERTs), and Amazon SNS (alerting for DELETEs)." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/dsql-cdc-eventbridge-fanout-cdk", + "templateURL": "serverless-patterns/dsql-cdc-eventbridge-fanout-cdk", + "projectFolder": "dsql-cdc-eventbridge-fanout-cdk", + "templateFile": "cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts" + } + }, + "resources": { + "bullets": [ + { + "text": "Amazon Aurora DSQL CDC documentation", + "link": "https://docs.aws.amazon.com/aurora-dsql/latest/userguide/cdc-streams.html" + }, + { + "text": "Amazon EventBridge content-based filtering", + "link": "https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html" + }, + { + "text": "AWS CDK Custom Resources", + "link": "https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.custom_resources-readme.html" + } + ] + }, + "deploy": { + "text": [ + "cd dsql-cdc-eventbridge-fanout-cdk/cdk", + "npm install", + "npx cdk deploy --parameters DsqlClusterId=your-cluster-id" + ] + }, + "testing": { + "text": [ + "Insert data into your Amazon Aurora DSQL cluster to generate CDC events.", + "Check the Amazon SQS audit queue for all change events.", + "Check AWS Step Functions executions for INSERT validation workflows.", + "Subscribe to the Amazon SNS topic to receive DELETE alerts." + ] + }, + "cleanup": { + "text": [ + "npx cdk destroy" + ] + }, + "authors": [ + { + "name": "Nithin Chandran R", + "image": "", + "bio": "Technical Account Manager at AWS", + "linkedin": "nithin-chandran-r", + "twitter": "" + } + ], + "patternArch": { + "icon1": { + "x": 10, + "y": 50, + "service": "dsql", + "label": "Amazon Aurora DSQL" + }, + "icon2": { + "x": 30, + "y": 50, + "service": "kinesis", + "label": "Amazon Kinesis" + }, + "icon3": { + "x": 50, + "y": 50, + "service": "lambda", + "label": "AWS Lambda" + }, + "icon4": { + "x": 70, + "y": 50, + "service": "eventbridge", + "label": "Amazon EventBridge" + }, + "icon5": { + "x": 90, + "y": 25, + "service": "sqs", + "label": "Amazon SQS" + }, + "icon6": { + "x": 90, + "y": 50, + "service": "sfn", + "label": "AWS Step Functions" + }, + "icon7": { + "x": 90, + "y": 75, + "service": "sns", + "label": "Amazon SNS" + } + }, + "patternType": "Serverless" +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py new file mode 100644 index 0000000000..43b6e878f5 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py @@ -0,0 +1,78 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: MIT-0 (2026) +"""CDC Processor: Parse Amazon Aurora DSQL CDC events from Amazon Kinesis +and fan out to Amazon EventBridge with typed detail-type routing.""" + +import base64 +import json +import os +from datetime import datetime, timezone + +import boto3 + +EVENTS_CLIENT = boto3.client('events') +EVENT_BUS_NAME = os.environ['EVENT_BUS_NAME'] +SOURCE = 'dsql.cdc' + + +def lambda_handler(event, context): + """Process Amazon Kinesis records containing Amazon Aurora DSQL CDC events.""" + records = event.get('Records', []) + if not records: + return {'statusCode': 200, 'processed': 0} + + entries = [] + for record in records: + try: + payload = base64.b64decode(record['kinesis']['data']) + cdc_event = json.loads(payload) + + # DSQL CDC JSON format contains: operation, tableName, newImage, oldImage + operation = cdc_event.get('operation', 'UNKNOWN').upper() + table_name = cdc_event.get('tableName', 'unknown') + + # Map to EventBridge detail-type for routing + detail_type = operation # INSERT, UPDATE, DELETE + + entry = { + 'Source': SOURCE, + 'DetailType': detail_type, + 'Detail': json.dumps({ + 'tableName': table_name, + 'operation': operation, + 'newImage': cdc_event.get('newImage'), + 'oldImage': cdc_event.get('oldImage'), + 'timestamp': cdc_event.get('commitTimestamp', + datetime.now(timezone.utc).isoformat()), + 'sequenceNumber': record['kinesis'].get('sequenceNumber'), + }), + 'EventBusName': EVENT_BUS_NAME, + } + entries.append(entry) + + except (json.JSONDecodeError, KeyError) as e: + print(f'Failed to parse CDC record: {e}') + continue + + # Publish in batches of 10 (EventBridge PutEvents limit) + failed_count = 0 + for i in range(0, len(entries), 10): + batch = entries[i:i + 10] + try: + response = EVENTS_CLIENT.put_events(Entries=batch) + failed_count += response.get('FailedEntryCount', 0) + except Exception as e: + print(f'EventBridge PutEvents error: {e}') + failed_count += len(batch) + + print(f'Processed {len(records)} records, published {len(entries)} events, ' + f'{failed_count} failures') + + if failed_count > 0: + raise Exception(f'{failed_count} events failed to publish to Amazon EventBridge') + + return { + 'statusCode': 200, + 'processed': len(records), + 'published': len(entries), + } diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py new file mode 100644 index 0000000000..4e0d0e9b11 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py @@ -0,0 +1,108 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: MIT-0 (2026) +"""Custom Resource: Create and delete Amazon Aurora DSQL CDC streams.""" + +import json +import os +import time + +import boto3 + +DSQL_CLIENT = boto3.client('dsql') + + +def on_event(event, context): + """Handle CloudFormation Custom Resource lifecycle events.""" + request_type = event['RequestType'] + properties = event['ResourceProperties'] + + cluster_id = properties.get('ClusterId', os.environ.get('CLUSTER_ID', '')) + kinesis_arn = properties.get('KinesisStreamArn', os.environ.get('KINESIS_STREAM_ARN', '')) + role_arn = properties.get('RoleArn', os.environ.get('CDC_ROLE_ARN', '')) + + if request_type == 'Create': + return create_stream(cluster_id, kinesis_arn, role_arn) + elif request_type == 'Update': + # CDC streams are immutable — delete old, create new + old_stream_id = event.get('PhysicalResourceId', '') + if old_stream_id: + delete_stream(cluster_id, old_stream_id) + return create_stream(cluster_id, kinesis_arn, role_arn) + elif request_type == 'Delete': + stream_id = event.get('PhysicalResourceId', '') + if stream_id and stream_id != 'NONE': + delete_stream(cluster_id, stream_id) + return {'PhysicalResourceId': stream_id or 'NONE'} + + +def create_stream(cluster_id: str, kinesis_arn: str, role_arn: str) -> dict: + """Create a DSQL CDC stream targeting Amazon Kinesis.""" + try: + response = DSQL_CLIENT.create_stream( + clusterIdentifier=cluster_id, + targetDefinition={ + 'kinesis': { + 'streamArn': kinesis_arn, + 'roleArn': role_arn, + } + }, + ordering='UNORDERED', + format='JSON', + ) + + stream_id = response['streamIdentifier'] + print(f'Created CDC stream: {stream_id} (status: {response["status"]})') + + # Wait for stream to become ACTIVE (max 60s) + wait_for_active(cluster_id, stream_id) + + return { + 'PhysicalResourceId': stream_id, + 'Data': { + 'StreamId': stream_id, + 'StreamArn': response.get('arn', ''), + 'Status': 'ACTIVE', + }, + } + + except Exception as e: + print(f'Failed to create CDC stream: {e}') + raise + + +def delete_stream(cluster_id: str, stream_id: str) -> None: + """Delete a DSQL CDC stream.""" + try: + DSQL_CLIENT.delete_stream( + clusterIdentifier=cluster_id, + streamIdentifier=stream_id, + ) + print(f'Deleted CDC stream: {stream_id}') + except DSQL_CLIENT.exceptions.ResourceNotFoundException: + print(f'Stream {stream_id} already deleted') + except Exception as e: + print(f'Failed to delete CDC stream {stream_id}: {e}') + raise + + +def wait_for_active(cluster_id: str, stream_id: str, max_wait: int = 60) -> None: + """Poll until stream status is ACTIVE.""" + waited = 0 + while waited < max_wait: + try: + response = DSQL_CLIENT.get_stream( + clusterIdentifier=cluster_id, + streamIdentifier=stream_id, + ) + status = response.get('status', '') + if status == 'ACTIVE': + return + if status == 'FAILED': + raise Exception(f'CDC stream {stream_id} entered FAILED state') + except Exception as e: + if 'ResourceNotFound' not in str(e): + raise + time.sleep(5) + waited += 5 + + print(f'Warning: stream {stream_id} not ACTIVE after {max_wait}s, proceeding') From 603977a8dc3f74507cb52b61e58d2caf31088429 Mon Sep 17 00:00:00 2001 From: Nithin Chandran Rajashankar Date: Mon, 13 Jul 2026 06:31:52 +0000 Subject: [PATCH 2/2] fix(dsql-cdc-eventbridge-fanout-cdk): Map DSQL CDC op codes and bundle boto3 DSQL CDC uses Debezium-style format: op='c'/'u'/'d' with before/after fields. Map op codes to human-readable types for EventBridge routing. Bundle boto3>=1.43.36 via CDK Docker bundling (Lambda runtime ships older version without dsql.create_stream support). Deployed and tested end-to-end: - 3 records processed, 3 events published, 0 failures - Amazon SQS audit: receives all 3 event types (INSERT, UPDATE, DELETE) - AWS Step Functions: 1 INSERT validation SUCCEEDED - Amazon SNS: DELETE event routed to alerting topic --- .../lib/dsql-cdc-eventbridge-fanout-stack.ts | 11 ++++++++-- .../lambdas/cdc-processor/handler.py | 21 ++++++++++++------- .../lambdas/cdc-stream-manager/.gitignore | 10 +++++++++ .../cdc-stream-manager/requirements.txt | 1 + 4 files changed, 33 insertions(+), 10 deletions(-) create mode 100644 dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/.gitignore create mode 100644 dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts b/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts index d12cc6ee30..e78cdb14ba 100644 --- a/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts @@ -9,7 +9,6 @@ import * as events from 'aws-cdk-lib/aws-events'; import * as targets from 'aws-cdk-lib/aws-events-targets'; import * as sqs from 'aws-cdk-lib/aws-sqs'; import * as sns from 'aws-cdk-lib/aws-sns'; -import * as snsSubs from 'aws-cdk-lib/aws-sns-subscriptions'; import * as sfn from 'aws-cdk-lib/aws-stepfunctions'; import * as sfnTasks from 'aws-cdk-lib/aws-stepfunctions-tasks'; import * as logs from 'aws-cdk-lib/aws-logs'; @@ -71,7 +70,15 @@ export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { const streamManagerFn = new lambda.Function(this, 'CdcStreamManagerFn', { runtime: lambda.Runtime.PYTHON_3_12, handler: 'handler.on_event', - code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/cdc-stream-manager')), + code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/cdc-stream-manager'), { + bundling: { + image: lambda.Runtime.PYTHON_3_12.bundlingImage, + command: [ + 'bash', '-c', + 'pip install -r requirements.txt -t /asset-output && cp handler.py /asset-output/', + ], + }, + }), timeout: cdk.Duration.minutes(5), memorySize: 256, description: 'Custom Resource: manages Amazon Aurora DSQL CDC stream lifecycle', diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py index 43b6e878f5..8d93c69596 100644 --- a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py @@ -21,18 +21,23 @@ def lambda_handler(event, context): if not records: return {'statusCode': 200, 'processed': 0} + # DSQL CDC JSON format: + # {"type":"full","op":"c|u|d","before":null|{...},"after":{...}|null,"source":{...}} + OP_MAP = {'c': 'INSERT', 'u': 'UPDATE', 'd': 'DELETE'} + entries = [] for record in records: try: payload = base64.b64decode(record['kinesis']['data']) cdc_event = json.loads(payload) - # DSQL CDC JSON format contains: operation, tableName, newImage, oldImage - operation = cdc_event.get('operation', 'UNKNOWN').upper() - table_name = cdc_event.get('tableName', 'unknown') + # Map DSQL CDC op codes to human-readable operation types + raw_op = cdc_event.get('op', '').lower() + operation = OP_MAP.get(raw_op, 'UNKNOWN') + table_name = cdc_event.get('source', {}).get('table', 'unknown') - # Map to EventBridge detail-type for routing - detail_type = operation # INSERT, UPDATE, DELETE + # Use operation as EventBridge detail-type for content-based routing + detail_type = operation entry = { 'Source': SOURCE, @@ -40,9 +45,9 @@ def lambda_handler(event, context): 'Detail': json.dumps({ 'tableName': table_name, 'operation': operation, - 'newImage': cdc_event.get('newImage'), - 'oldImage': cdc_event.get('oldImage'), - 'timestamp': cdc_event.get('commitTimestamp', + 'newImage': cdc_event.get('after'), + 'oldImage': cdc_event.get('before'), + 'timestamp': cdc_event.get('source', {}).get('ts_ms', datetime.now(timezone.utc).isoformat()), 'sequenceNumber': record['kinesis'].get('sequenceNumber'), }), diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/.gitignore b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/.gitignore new file mode 100644 index 0000000000..61f07b0254 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/.gitignore @@ -0,0 +1,10 @@ +# Vendored dependencies (bundled at deploy time, not committed) +boto3/ +botocore/ +s3transfer/ +urllib3/ +jmespath/ +dateutil/ +six.py +*.dist-info/ +bin/ diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt new file mode 100644 index 0000000000..c26e4b5794 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt @@ -0,0 +1 @@ +boto3>=1.43.36