From 74d625fca2dae4911ed5568ea385230a3819ed63 Mon Sep 17 00:00:00 2001 From: lucas Date: Tue, 30 Jun 2026 01:06:43 +0000 Subject: [PATCH] feat(compute): support SQS provisioned pollers --- src/aws/compute/event-source-mapping.ts | 68 +++++++++++++ src/aws/compute/event-sources/sqs.ts | 42 +++++++++ test/aws/compute/event-sources/sqs.test.ts | 105 +++++++++++++++++++++ 3 files changed, 215 insertions(+) diff --git a/src/aws/compute/event-source-mapping.ts b/src/aws/compute/event-source-mapping.ts index ba8dc993..36c8688a 100644 --- a/src/aws/compute/event-source-mapping.ts +++ b/src/aws/compute/event-source-mapping.ts @@ -15,6 +15,38 @@ import { withResolved } from "../../token"; import { IKey } from "../encryption"; import * as iam from "../iam"; +export interface ProvisionedPollerConfig { + /** + * The minimum number of pollers that should be provisioned. + * + * Valid Range: + * * For Amazon SQS: Minimum value of 2. Maximum value of 200. Default: 2. + * * For Amazon MSK, self-managed Apache Kafka, and Amazon MQ: Minimum value of 1. Maximum value of 200. Default: 1. + * + * @default - 2 for SQS, 1 for MSK/Kafka/MQ + */ + readonly minimumPollers?: number; + + /** + * The maximum number of pollers that can be provisioned. + * + * Valid Range: + * * For Amazon SQS: Minimum value of 2. Maximum value of 2000. Default: 200. + * * For Amazon MSK, self-managed Apache Kafka, and Amazon MQ: Minimum value of 1. Maximum value of 2000. Default: 200. + * + * @default 200 + */ + readonly maximumPollers?: number; + + /** + * An optional identifier that groups multiple event source mappings to share + * poller capacity and reduce costs. + * + * @default - not set, dedicated compute resource per event source. + */ + readonly pollerGroupName?: string; +} + export interface EventSourceMappingOptions extends AwsConstructProps { /** * The Amazon Resource Name (ARN) of the event source. Any record added to @@ -102,6 +134,15 @@ export interface EventSourceMappingOptions extends AwsConstructProps { */ readonly maxConcurrency?: number; + /** + * Configuration for provisioned pollers that read from the event source. + * When specified, allows control over the minimum and maximum number of + * pollers that can be provisioned to process events from the source. + * + * @default - no provisioned pollers + */ + readonly provisionedPollerConfig?: ProvisionedPollerConfig; + /** * The maximum age of a record that Lambda sends to a function for processing. * Valid Range: @@ -360,6 +401,32 @@ export class EventSourceMapping ); } + if (props.provisionedPollerConfig) { + const { minimumPollers, maximumPollers } = props.provisionedPollerConfig; + const hasMin = + minimumPollers !== undefined && + !cdktn.Token.isUnresolved(minimumPollers); + const hasMax = + maximumPollers !== undefined && + !cdktn.Token.isUnresolved(maximumPollers); + + if (hasMin && (minimumPollers < 1 || minimumPollers > 200)) { + throw new Error( + "Minimum provisioned pollers must be between 1 and 200 inclusive", + ); + } + if (hasMax && (maximumPollers < 1 || maximumPollers > 2000)) { + throw new Error( + "Maximum provisioned pollers must be between 1 and 2000 inclusive", + ); + } + if (hasMin && hasMax && minimumPollers > maximumPollers) { + throw new Error( + "Minimum provisioned pollers must be less than or equal to maximum provisioned pollers", + ); + } + } + if ( props.maxRecordAge && (props.maxRecordAge.toSeconds() < 60 || @@ -488,6 +555,7 @@ export class EventSourceMapping scalingConfig: props.maxConcurrency ? { maximumConcurrency: props.maxConcurrency } : undefined, + provisionedPollerConfig: props.provisionedPollerConfig, sourceAccessConfiguration: props.sourceAccessConfigurations, selfManagedEventSource, filterCriteria: props.filters ? { filter: props.filters } : undefined, diff --git a/src/aws/compute/event-sources/sqs.ts b/src/aws/compute/event-sources/sqs.ts index a4c0ac7f..c3fc9b91 100644 --- a/src/aws/compute/event-sources/sqs.ts +++ b/src/aws/compute/event-sources/sqs.ts @@ -72,6 +72,17 @@ export interface SqsEventSourceProps { * @default - No specific limit. */ readonly maxConcurrency?: number; + + /** + * Configuration for provisioned pollers that read from the event source. + * When specified, allows control over the minimum and maximum number of + * pollers that can be provisioned to process events from the queue. + * + * @see https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html + * + * @default - no provisioned pollers + */ + readonly provisionedPollerConfig?: compute.ProvisionedPollerConfig; } /** @@ -119,6 +130,36 @@ export class SqsEventSource implements compute.IEventSource { ); } } + if (this.props.provisionedPollerConfig) { + if (this.props.maxConcurrency !== undefined) { + throw new Error( + "provisionedPollerConfig and maxConcurrency are mutually exclusive — specify only one", + ); + } + + const { minimumPollers, maximumPollers } = + this.props.provisionedPollerConfig; + const hasMin = + minimumPollers !== undefined && !Token.isUnresolved(minimumPollers); + const hasMax = + maximumPollers !== undefined && !Token.isUnresolved(maximumPollers); + + if (hasMin && (minimumPollers < 2 || minimumPollers > 200)) { + throw new Error( + `Minimum provisioned pollers for SQS must be between 2 and 200 inclusive, got: ${minimumPollers}`, + ); + } + if (hasMax && (maximumPollers < 2 || maximumPollers > 2000)) { + throw new Error( + `Maximum provisioned pollers for SQS must be between 2 and 2000 inclusive, got: ${maximumPollers}`, + ); + } + if (hasMin && hasMax && minimumPollers > maximumPollers) { + throw new Error( + `Minimum provisioned pollers must be less than or equal to maximum provisioned pollers, got: min=${minimumPollers}, max=${maximumPollers}`, + ); + } + } } public bind(target: compute.IFunction) { @@ -133,6 +174,7 @@ export class SqsEventSource implements compute.IEventSource { eventSourceArn: this.queue.queueArn, filters: this.props.filters, filterEncryption: this.props.filterEncryption, + provisionedPollerConfig: this.props.provisionedPollerConfig, }, ); this._eventSourceMappingId = eventSourceMapping.eventSourceMappingId; diff --git a/test/aws/compute/event-sources/sqs.test.ts b/test/aws/compute/event-sources/sqs.test.ts index 2ccad5f1..4b16b395 100644 --- a/test/aws/compute/event-sources/sqs.test.ts +++ b/test/aws/compute/event-sources/sqs.test.ts @@ -629,6 +629,111 @@ describe("SQSEventSource", () => { ); }); + test("adding provisionedPollerConfig", () => { + // GIVEN + const fn = new TestFunction(stack, "Fn"); + const q = new notify.Queue(stack, "Q"); + + // WHEN + fn.addEventSource( + new compute.sources.SqsEventSource(q, { + provisionedPollerConfig: { + minimumPollers: 2, + maximumPollers: 10, + }, + }), + ); + + // THEN + Template.synth(stack).toHaveResourceWithProperties( + lambdaEventSourceMapping.LambdaEventSourceMapping, + { + provisioned_poller_config: { + minimum_pollers: 2, + maximum_pollers: 10, + }, + }, + ); + }); + + test.each([1, 201])( + "fails if minimumPollers for SQS is out of range (%i)", + (minimumPollers) => { + // GIVEN + const fn = new TestFunction(stack, "Fn"); + const q = new notify.Queue(stack, "Q"); + + // WHEN/THEN + expect(() => + fn.addEventSource( + new compute.sources.SqsEventSource(q, { + provisionedPollerConfig: { minimumPollers }, + }), + ), + ).toThrow( + `Minimum provisioned pollers for SQS must be between 2 and 200 inclusive, got: ${minimumPollers}`, + ); + }, + ); + + test.each([1, 2001])( + "fails if maximumPollers for SQS is out of range (%i)", + (maximumPollers) => { + // GIVEN + const fn = new TestFunction(stack, "Fn"); + const q = new notify.Queue(stack, "Q"); + + // WHEN/THEN + expect(() => + fn.addEventSource( + new compute.sources.SqsEventSource(q, { + provisionedPollerConfig: { maximumPollers }, + }), + ), + ).toThrow( + `Maximum provisioned pollers for SQS must be between 2 and 2000 inclusive, got: ${maximumPollers}`, + ); + }, + ); + + test("fails if minimumPollers exceeds maximumPollers for SQS", () => { + // GIVEN + const fn = new TestFunction(stack, "Fn"); + const q = new notify.Queue(stack, "Q"); + + // WHEN/THEN + expect(() => + fn.addEventSource( + new compute.sources.SqsEventSource(q, { + provisionedPollerConfig: { + minimumPollers: 10, + maximumPollers: 2, + }, + }), + ), + ).toThrow( + "Minimum provisioned pollers must be less than or equal to maximum provisioned pollers, got: min=10, max=2", + ); + }); + + test("fails if provisionedPollerConfig and maxConcurrency are specified together", () => { + // GIVEN + const fn = new TestFunction(stack, "Fn"); + const q = new notify.Queue(stack, "Q"); + + // WHEN/THEN + expect(() => + fn.addEventSource( + new compute.sources.SqsEventSource(q, { + maxConcurrency: 5, + provisionedPollerConfig: { minimumPollers: 2, maximumPollers: 10 }, + }), + ), + ).toThrow( + "provisionedPollerConfig and maxConcurrency are mutually exclusive — specify only one", + ); + }); + test("fails if maxConcurrency < 2", () => { // GIVEN const fn = new TestFunction(stack, "Fn");