feat(compute): support SQS provisioned pollers

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,

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;

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");