feat(lambda-event-sources): tumbling window

packages/@aws-cdk/aws-lambda-event-sources/README.md

@@ -148,6 +148,7 @@ and add it to your Lambda function. The following parameters will impact Amazon
 * __parallelizationFactor__: The number of batches to concurrently process on each shard.
 * __retryAttempts__: The maximum number of times a record should be retried in the event of failure.
 * __startingPosition__: Will determine where to being consumption, either at the most recent ('LATEST') record or the oldest record ('TRIM_HORIZON'). 'TRIM_HORIZON' will ensure you process all available data, while 'LATEST' will ignore all records that arrived prior to attaching the event source.
+* __tumblingWindow__: The duration in seconds of a processing window when using streams.
 * __enabled__: If the DynamoDB Streams event source mapping should be enabled. The default is true.
 
 ```ts
@@ -192,6 +193,7 @@ behavior:
 * __parallelizationFactor__: The number of batches to concurrently process on each shard.
 * __retryAttempts__: The maximum number of times a record should be retried in the event of failure.
 * __startingPosition__: Will determine where to being consumption, either at the most recent ('LATEST') record or the oldest record ('TRIM_HORIZON'). 'TRIM_HORIZON' will ensure you process all available data, while 'LATEST' will ignore all records that arrived prior to attaching the event source.
+* __tumblingWindow__: The duration in seconds of a processing window when using streams.
 * __enabled__: If the DynamoDB Streams event source mapping should be enabled. The default is true.
 
 ```ts

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

@@ -78,6 +78,16 @@ export interface StreamEventSourceProps {
    */
   readonly maxBatchingWindow?: Duration;
 
+  /**
+   * The size of the tumbling windows to group records sent to DynamoDB or Kinesis
+   * Valid Range:
+   * * Minimum value of 0
+   * * Maximum value of 900
+   *
+   * @default - None
+   */
+  readonly tumblingWindow?: Duration;
+
   /**
    * If the stream event source mapping should be enabled.
    *
@@ -106,6 +116,7 @@ export abstract class StreamEventSource implements lambda.IEventSource {
       retryAttempts: this.props.retryAttempts,
       parallelizationFactor: this.props.parallelizationFactor,
       onFailure: this.props.onFailure,
+      tumblingWindow: this.props.tumblingWindow,
       enabled: this.props.enabled,
     };
   }

packages/@aws-cdk/aws-lambda-event-sources/test/integ.dynamodb.expected.json

@@ -99,7 +99,8 @@
           "Ref": "FC4345940"
         },
         "BatchSize": 5,
-        "StartingPosition": "TRIM_HORIZON"
+        "StartingPosition": "TRIM_HORIZON",
+        "TumblingWindowInSeconds": 60
       }
     },
     "TD925BC7E": {

packages/@aws-cdk/aws-lambda-event-sources/test/integ.dynamodb.ts

@@ -21,6 +21,7 @@ class DynamoEventSourceTest extends cdk.Stack {
     fn.addEventSource(new DynamoEventSource(queue, {
       batchSize: 5,
       startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+      tumblingWindow: cdk.Duration.seconds(60),
     }));
   }
 }

packages/@aws-cdk/aws-lambda-event-sources/test/integ.kinesis.expected.json

@@ -106,7 +106,8 @@
           "Ref": "FC4345940"
         },
         "BatchSize": 100,
-        "StartingPosition": "TRIM_HORIZON"
+        "StartingPosition": "TRIM_HORIZON",
+        "TumblingWindowInSeconds": 60
       }
     },
     "Q63C6E3AB": {

packages/@aws-cdk/aws-lambda-event-sources/test/integ.kinesis.ts

@@ -13,6 +13,7 @@ class KinesisEventSourceTest extends cdk.Stack {
 
     fn.addEventSource(new KinesisEventSource(stream, {
       startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+      tumblingWindow: cdk.Duration.seconds(60),
     }));
   }
 }

packages/@aws-cdk/aws-lambda-event-sources/test/test.dynamo.ts

@@ -76,6 +76,67 @@ export = {
     test.done();
   },
 
+  'specific tumblingWindow'(test: Test) {
+    // GIVEN
+    const stack = new cdk.Stack();
+    const fn = new TestFunction(stack, 'Fn');
+    const table = new dynamodb.Table(stack, 'T', {
+      partitionKey: {
+        name: 'id',
+        type: dynamodb.AttributeType.STRING,
+      },
+      stream: dynamodb.StreamViewType.NEW_IMAGE,
+    });
+
+    // WHEN
+    fn.addEventSource(new sources.DynamoEventSource(table, {
+      batchSize: 50,
+      startingPosition: lambda.StartingPosition.LATEST,
+      tumblingWindow: cdk.Duration.seconds(60),
+    }));
+
+    // THEN
+    expect(stack).to(haveResource('AWS::Lambda::EventSourceMapping', {
+      'EventSourceArn': {
+        'Fn::GetAtt': [
+          'TD925BC7E',
+          'StreamArn',
+        ],
+      },
+      'FunctionName': {
+        'Ref': 'Fn9270CBC0',
+      },
+      'BatchSize': 50,
+      'StartingPosition': 'LATEST',
+      'TumblingWindowInSeconds': 60,
+    }));
+
+    test.done();
+  },
+
+  'fails if tumblingWindow > 900 seconds'(test: Test) {
+    // GIVEN
+    const stack = new cdk.Stack();
+    const fn = new TestFunction(stack, 'Fn');
+    const table = new dynamodb.Table(stack, 'T', {
+      partitionKey: {
+        name: 'id',
+        type: dynamodb.AttributeType.STRING,
+      },
+      stream: dynamodb.StreamViewType.NEW_IMAGE,
+    });
+
+    // THEN
+    test.throws(() =>
+      fn.addEventSource(new sources.DynamoEventSource(table, {
+        maxBatchingWindow: cdk.Duration.seconds(60),
+        startingPosition: lambda.StartingPosition.LATEST,
+        tumblingWindow: cdk.Duration.seconds(901),
+      })), /tumblingWindow cannot be over 900 seconds/);
+
+    test.done();
+  },
+
   'specific batch size'(test: Test) {
     // GIVEN
     const stack = new cdk.Stack();

packages/@aws-cdk/aws-lambda-event-sources/test/test.kinesis.ts

@@ -76,6 +76,55 @@ export = {
     test.done();
   },
 
+  'specific tumblingWindowInSeconds'(test: Test) {
+    // GIVEN
+    const stack = new cdk.Stack();
+    const fn = new TestFunction(stack, 'Fn');
+    const stream = new kinesis.Stream(stack, 'S');
+
+    // WHEN
+    fn.addEventSource(new sources.KinesisEventSource(stream, {
+      batchSize: 50,
+      startingPosition: lambda.StartingPosition.LATEST,
+      tumblingWindow: cdk.Duration.seconds(60),
+    }));
+
+    // THEN
+    expect(stack).to(haveResource('AWS::Lambda::EventSourceMapping', {
+      'EventSourceArn': {
+        'Fn::GetAtt': [
+          'S509448A1',
+          'Arn',
+        ],
+      },
+      'FunctionName': {
+        'Ref': 'Fn9270CBC0',
+      },
+      'BatchSize': 50,
+      'StartingPosition': 'LATEST',
+      'TumblingWindowInSeconds': 60,
+    }));
+
+    test.done();
+  },
+
+  'fails if tumblingWindow > 900 seconds'(test: Test) {
+    // GIVEN
+    const stack = new cdk.Stack();
+    const fn = new TestFunction(stack, 'Fn');
+    const stream = new kinesis.Stream(stack, 'S');
+
+    // THEN
+    test.throws(() =>
+      fn.addEventSource(new sources.KinesisEventSource(stream, {
+        maxBatchingWindow: cdk.Duration.seconds(60),
+        startingPosition: lambda.StartingPosition.LATEST,
+        tumblingWindow: cdk.Duration.seconds(901),
+      })), /tumblingWindow cannot be over 900 seconds/);
+
+    test.done();
+  },
+
   'specific batch size'(test: Test) {
     // GIVEN
     const stack = new cdk.Stack();

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

@@ -101,6 +101,19 @@ export interface EventSourceMappingOptions {
    * @default - no topic
    */
   readonly kafkaTopic?: string;
+
+  /**
+   * The size of the tumbling windows to group records sent to DynamoDB or Kinesis
+   *
+   * @see https://docs.aws.amazon.com/lambda/latest/dg/with-ddb.html#services-ddb-windows
+   *
+   * Valid Range:
+   * * Minimum value of 0
+   * * Maximum value of 900
+   *
+   * @default - None
+   */
+  readonly tumblingWindow?: cdk.Duration;
 }
 
 /**
@@ -174,6 +187,10 @@ export class EventSourceMapping extends cdk.Resource implements IEventSourceMapp
       }
     });
 
+    if (props.tumblingWindow && props.tumblingWindow.toSeconds() > 900) {
+      throw new Error(`tumblingWindow cannot be over 900 seconds, got ${props.tumblingWindow.toSeconds()}`);
+    }
+
 
     let destinationConfig;
 
@@ -196,6 +213,7 @@ export class EventSourceMapping extends cdk.Resource implements IEventSourceMapp
       maximumRetryAttempts: props.retryAttempts,
       parallelizationFactor: props.parallelizationFactor,
       topics: props.kafkaTopic !== undefined ? [props.kafkaTopic] : undefined,
+      tumblingWindowInSeconds: props.tumblingWindow?.toSeconds(),
     });
     this.eventSourceMappingId = cfnEventSourceMapping.ref;
   }

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

@@ -170,4 +170,19 @@ describe('event source mapping', () => {
       }],
     });
   });
+
+  test('throws if tumblingWindow > 900 seconds', () => {
+    const stack = new cdk.Stack();
+    const fn = new Function(stack, 'fn', {
+      handler: 'index.handler',
+      code: Code.fromInline('exports.handler = ${handler.toString()}'),
+      runtime: Runtime.NODEJS_10_X,
+    });
+
+    expect(() => new EventSourceMapping(stack, 'test', {
+      target: fn,
+      eventSourceArn: '',
+      tumblingWindow: cdk.Duration.seconds(901),
+    })).toThrow(/tumblingWindow cannot be over 900 seconds/);
+  });
 });