fix(lambda): add Token.isUnresolved checks to provisioned poller config validation (#37197)

### Issue # (if applicable)

None

### Reason for this change

The `provisionedPollerConfig` validation in both `EventSourceMapping` (aws-lambda) and `StreamEventSource` (aws-lambda-event-sources) throws errors when unresolved CDK tokens are passed as `minimumPollers` or `maximumPollers` values (e.g. `CfnParameter`, `Lazy.number()`). These values cannot be validated at synthesis time and should be skipped, as is already done for other numeric validations like `maxConcurrency` and `tumblingWindow`.

### Description of changes

Added `Token.isUnresolved()` guards to provisioned poller config validation in:

- `packages/aws-cdk-lib/aws-lambda/lib/event-source-mapping.ts`
- `packages/aws-cdk-lib/aws-lambda-event-sources/lib/stream.ts`

This follows the same pattern already used for `maxConcurrency` (line 542 of event-source-mapping.ts) and `tumblingWindow` (line 563).

### Describe any new or updated permissions being added

None

### Description of how you validated changes

Add unit tests.

### Checklist
- [x] My code adheres to the [CONTRIBUTING GUIDE](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md) and [DESIGN GUIDELINES](https://github.com/aws/aws-cdk/blob/main/docs/DESIGN_GUIDELINES.md)

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: badmintoncryer

packages/aws-cdk-lib/aws-lambda-event-sources/lib/stream.ts

@@ -2,7 +2,7 @@ import { S3OnFailureDestination } from './s3-onfailuire-destination';
 import type { IKey } from '../../aws-kms';
 import type * as lambda from '../../aws-lambda';
 import type { Duration } from '../../core';
-import { UnscopedValidationError } from '../../core';
+import { Token, UnscopedValidationError } from '../../core';
 import { lit } from '../../core/lib/private/literal-string';
 
 /**
@@ -186,17 +186,19 @@ export abstract class StreamEventSource implements lambda.IEventSource {
   protected constructor(protected readonly props: StreamEventSourceProps) {
     if (props.provisionedPollerConfig) {
       const { minimumPollers, maximumPollers } = props.provisionedPollerConfig;
-      if (minimumPollers != undefined) {
+      const isMinimumPollersDefinedAndResolved = minimumPollers != undefined && !Token.isUnresolved(minimumPollers);
+      const isMaximumPollersDefinedAndResolved = maximumPollers != undefined && !Token.isUnresolved(maximumPollers);
+      if (isMinimumPollersDefinedAndResolved) {
         if (minimumPollers < 1 || minimumPollers > 200) {
           throw new UnscopedValidationError(lit`MustBeMinimumProvisionedPollers`, 'Minimum provisioned pollers must be between 1 and 200 inclusive');
         }
       }
-      if (maximumPollers != undefined) {
+      if (isMaximumPollersDefinedAndResolved) {
         if (maximumPollers < 1 || maximumPollers > 2000) {
           throw new UnscopedValidationError(lit`MustBeMaximumProvisionedPollers`, 'Maximum provisioned pollers must be between 1 and 2000 inclusive');
         }
       }
-      if (minimumPollers != undefined && maximumPollers != undefined) {
+      if (isMinimumPollersDefinedAndResolved && isMaximumPollersDefinedAndResolved) {
         if (minimumPollers > maximumPollers) {
           throw new UnscopedValidationError(lit`MustBeMinimumProvisionedPollers`, 'Minimum provisioned pollers must be less than or equal to maximum provisioned pollers');
         }

packages/aws-cdk-lib/aws-lambda-event-sources/test/kafka.test.ts

@@ -577,6 +577,27 @@ describe('KafkaEventSource', () => {
         }))).toThrow(/Minimum provisioned pollers must be less than or equal to maximum provisioned pollers/);
     });
 
+    test('provisioned pollers with unresolved tokens should not throw', () => {
+      const stack = new cdk.Stack();
+      const testLambdaFunction = new TestFunction(stack, 'Fn');
+      const clusterArn = 'some-arn';
+      const kafkaTopic = 'some-topic';
+      const bucket = Bucket.fromBucketName(stack, 'BucketByName', 'my-bucket');
+      const s3OnFailureDestination = new sources.S3OnFailureDestination(bucket);
+
+      expect(() => testLambdaFunction.addEventSource(new sources.ManagedKafkaEventSource(
+        {
+          clusterArn,
+          topic: kafkaTopic,
+          startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+          onFailure: s3OnFailureDestination,
+          provisionedPollerConfig: {
+            minimumPollers: cdk.Lazy.number({ produce: () => 1 }),
+            maximumPollers: cdk.Lazy.number({ produce: () => 3 }),
+          },
+        }))).not.toThrow();
+    });
+
     test('MetricsConfig validation - empty metrics array', () => {
       // GIVEN
       const stack = new cdk.Stack();
@@ -1439,6 +1460,25 @@ describe('KafkaEventSource', () => {
         }))).toThrow(/Minimum provisioned pollers must be less than or equal to maximum provisioned pollers/);
     });
 
+    test('provisioned pollers with unresolved tokens should not throw', () => {
+      // GIVEN
+      const stack = new cdk.Stack();
+      const fn = new TestFunction(stack, 'Fn');
+      const clusterArn = 'some-arn';
+      const kafkaTopic = 'some-topic';
+
+      expect(() => fn.addEventSource(new sources.ManagedKafkaEventSource(
+        {
+          clusterArn,
+          topic: kafkaTopic,
+          startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+          provisionedPollerConfig: {
+            minimumPollers: cdk.Lazy.number({ produce: () => 1 }),
+            maximumPollers: cdk.Lazy.number({ produce: () => 3 }),
+          },
+        }))).not.toThrow();
+    });
+
     test('MetricsConfig validation - empty metrics array', () => {
       // GIVEN
       const stack = new cdk.Stack();

packages/aws-cdk-lib/aws-lambda/lib/event-source-mapping.ts

@@ -515,17 +515,19 @@ export class EventSourceMapping extends cdk.Resource implements IEventSourceMapp
 
     if (props.provisionedPollerConfig) {
       const { minimumPollers, maximumPollers } = props.provisionedPollerConfig;
-      if (minimumPollers != undefined) {
+      const isMinimumPollersDefinedAndResolved = minimumPollers != undefined && !cdk.Token.isUnresolved(minimumPollers);
+      const isMaximumPollersDefinedAndResolved = maximumPollers != undefined && !cdk.Token.isUnresolved(maximumPollers);
+      if (isMinimumPollersDefinedAndResolved) {
         if (minimumPollers < 1 || minimumPollers > 200) {
           throw new ValidationError(lit`MinimumProvisionedPollersInclusive`, 'Minimum provisioned pollers must be between 1 and 200 inclusive', this);
         }
       }
-      if (maximumPollers != undefined) {
+      if (isMaximumPollersDefinedAndResolved) {
         if (maximumPollers < 1 || maximumPollers > 2000) {
           throw new ValidationError(lit`MaximumProvisionedPollersInclusive`, 'Maximum provisioned pollers must be between 1 and 2000 inclusive', this);
         }
       }
-      if (minimumPollers != undefined && maximumPollers != undefined) {
+      if (isMinimumPollersDefinedAndResolved && isMaximumPollersDefinedAndResolved) {
         if (minimumPollers > maximumPollers) {
           throw new ValidationError(lit`MinimumProvisionedPollersLessEqual`, 'Minimum provisioned pollers must be less than or equal to maximum provisioned pollers', this);
         }

packages/aws-cdk-lib/aws-lambda/test/event-source-mapping.test.ts

@@ -602,4 +602,17 @@ describe('event source mapping', () => {
       },
     })).toThrow(/Minimum provisioned pollers must be less than or equal to maximum provisioned pollers/);
   });
+
+  test('provisioned pollers with unresolved tokens should not throw', () => {
+    expect(() => new EventSourceMapping(stack, 'test', {
+      target: fn,
+      eventSourceArn: '',
+      startingPosition: StartingPosition.AT_TIMESTAMP,
+      startingPositionTimestamp: 1640995200,
+      provisionedPollerConfig: {
+        minimumPollers: cdk.Lazy.number({ produce: () => 1 }),
+        maximumPollers: cdk.Lazy.number({ produce: () => 3 }),
+      },
+    })).not.toThrow();
+  });
 });