feat(lambda): event-source maxBatchingWindow property

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

@@ -1,5 +1,6 @@
 import dynamodb = require('@aws-cdk/aws-dynamodb');
 import lambda = require('@aws-cdk/aws-lambda');
+import { Duration } from "@aws-cdk/core";
 
 export interface DynamoEventSourceProps {
   /**
@@ -17,6 +18,14 @@ export interface DynamoEventSourceProps {
    * Where to begin consuming the DynamoDB stream.
    */
   readonly startingPosition: lambda.StartingPosition;
+
+  /**
+   * The maximum amount of time to gather records before invoking the function.
+   * Maximum of Duration.minutes(5)
+   *
+   * @default Duration.seconds(0)
+   */
+  readonly maximumBatchingWindow?: Duration;
 }
 
 /**
@@ -37,7 +46,8 @@ export class DynamoEventSource implements lambda.IEventSource {
     target.addEventSourceMapping(`DynamoDBEventSource:${this.table.node.uniqueId}`, {
       batchSize: this.props.batchSize || 100,
       eventSourceArn: this.table.tableStreamArn,
-      startingPosition: this.props.startingPosition
+      startingPosition: this.props.startingPosition,
+      maximumBatchingWindow: this.props.maximumBatchingWindow,
     });
 
     this.table.grantStreamRead(target);

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

@@ -1,5 +1,6 @@
 import kinesis = require('@aws-cdk/aws-kinesis');
 import lambda = require('@aws-cdk/aws-lambda');
+import {Duration} from "@aws-cdk/core";
 
 export interface KinesisEventSourceProps {
   /**
@@ -17,6 +18,14 @@ export interface KinesisEventSourceProps {
    * Where to begin consuming the Kinesis stream.
    */
   readonly startingPosition: lambda.StartingPosition;
+
+  /**
+   * The maximum amount of time to gather records before invoking the function.
+   * Maximum of Duration.minutes(5)
+   *
+   * @default Duration.seconds(0)
+   */
+  readonly maximumBatchingWindow?: Duration;
 }
 
 /**
@@ -34,6 +43,7 @@ export class KinesisEventSource implements lambda.IEventSource {
       batchSize: this.props.batchSize || 100,
       startingPosition: this.props.startingPosition,
       eventSourceArn: this.stream.streamArn,
+      maximumBatchingWindow: this.props.maximumBatchingWindow,
     });
 
     this.stream.grantRead(target);

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

@@ -172,4 +172,63 @@ export = {
 
     test.done();
   },
+
+  'specific maximumBatchingWindow'(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, {
+      maximumBatchingWindow: cdk.Duration.minutes(2),
+      startingPosition: lambda.StartingPosition.LATEST
+    }));
+
+    // THEN
+    expect(stack).to(haveResource('AWS::Lambda::EventSourceMapping', {
+      "EventSourceArn": {
+        "Fn::GetAtt": [
+          "TD925BC7E",
+          "StreamArn"
+        ]
+      },
+      "FunctionName":  {
+        "Ref": "Fn9270CBC0"
+      },
+      "MaximumBatchingWindowInSeconds": 120,
+      "StartingPosition": "LATEST"
+    }));
+
+    test.done();
+  },
+
+  'throws if maximumBatchingWindow > 300 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, {
+        maximumBatchingWindow: cdk.Duration.seconds(301),
+        startingPosition: lambda.StartingPosition.LATEST
+      })), /maximumBatchingWindow cannot be over 300 seconds/);
+
+    test.done();
+  },
+
 };

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

@@ -123,4 +123,34 @@ export = {
 
     test.done();
   },
+
+  'specific maximumBatchingWindow'(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, {
+      maximumBatchingWindow: cdk.Duration.minutes(2),
+      startingPosition: lambda.StartingPosition.LATEST
+    }));
+
+    // THEN
+    expect(stack).to(haveResource('AWS::Lambda::EventSourceMapping', {
+      "EventSourceArn": {
+        "Fn::GetAtt": [
+          "S509448A1",
+          "Arn"
+        ]
+      },
+      "FunctionName":  {
+        "Ref": "Fn9270CBC0"
+      },
+      "MaximumBatchingWindowInSeconds": 120,
+      "StartingPosition": "LATEST"
+    }));
+
+    test.done();
+  },
 };

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

@@ -37,7 +37,15 @@ export interface EventSourceMappingOptions {
    *
    * @default - Required for Amazon Kinesis and Amazon DynamoDB Streams sources.
    */
-  readonly startingPosition?: StartingPosition
+  readonly startingPosition?: StartingPosition;
+
+  /**
+   * The maximum amount of time to gather records before invoking the function.
+   * Maximum of Duration.minutes(5)
+   *
+   * @default Duration.seconds(0)
+   */
+  readonly maximumBatchingWindow?: cdk.Duration;
 }
 
 export interface EventSourceMappingProps extends EventSourceMappingOptions {
@@ -63,12 +71,17 @@ export class EventSourceMapping extends Resource {
   constructor(scope: cdk.Construct, id: string, props: EventSourceMappingProps) {
     super(scope, id);
 
+    if (props.maximumBatchingWindow && props.maximumBatchingWindow.toSeconds() > 300) {
+      throw new Error(`maximumBatchingWindow cannot be over 300 seconds, got ${props.maximumBatchingWindow.toSeconds()}`);
+    }
+
     new CfnEventSourceMapping(this, 'Resource', {
       batchSize: props.batchSize,
       enabled: props.enabled,
       eventSourceArn: props.eventSourceArn,
       functionName: props.target.functionName,
       startingPosition: props.startingPosition,
+      maximumBatchingWindowInSeconds: props.maximumBatchingWindow && props.maximumBatchingWindow.toSeconds(),
     });
   }
 }