Repository: localstack/localstack
Pull request: Make Lambda batch size configurable for Kinesis event source mappings
* Set default batch size event source = 10
* Create multiple events according to the batch size

Fixes #1131

Problem statement:
Lambda Event Source Mapping From Kinesis Does Not Always Honor Batch Size
I have a setup where there is a node.js application that writes to a kinesis stream using putRecords for two items. The kinesis stream feeds a Lambda function that is configured with an eventSourceMapping with a batch size of 1. However, when I debug what the Records Kinesis feeds the lambda for an invocation, I see that the lambda is processing two records. Both of these records are on the same shard and have a sequence number which is pretty close.

The interesting thing is that if I switch the node.js published to use putRecord instead of putRecords, the kinesis records have a slightly farther apart sequence number and the lambda is fed the records individually.

The java code to configure the event source mapping
```
        AWSLambdaAsync lambdaClient = AWSResourceFactory.getLambdaClient();
        CreateEventSourceMappingRequest request = new CreateEventSourceMappingRequest();
        request.setEventSourceArn(kinesisArn);
        request.setFunctionName(functionName);
        request.setEnabled(true);
        request.setStartingPosition(EventSourcePosition.TRIM_HORIZON);
        request.setBatchSize(1);
        lambdaClient.createEventSourceMapping(request);
```

The localstack lambda output that shows multiple records being read at a time:

```
{
	"Records": [{
		"eventID": "shardId-000000000000:49593014281924840790139112603493819506073986345293840386",
		"eventSourceARN": "arn:aws:kinesis:us-east-1:000000000000:stream/MyStream",
		"kinesis": {
			"partitionKey": "ed0f756b97234f4fae9e3d1a886b8fbc",
			"data": "myData1",
			"sequenceNumber": "49593014281924840790139112603493819506073986345293840386"
		}
	}, {
		"eventID": "shardId-000000000000:49593014281924840790139112603495028431893600974468546562",
		"eventSourceARN": "arn:aws:kinesis:us-east-1:000000000000:stream/MyStream",
		"kinesis": {
			"partitionKey": "abee63a9299385699d65e89a290a9065",
			"data": "myData2",
			"sequenceNumber": "49593014281924840790139112603495028431893600974468546562"
		}
	}]
}
```

Hints from issue discussion:
Hi, thanks for reporting @tmoser2525 . This should be relatively easy to achieve - instead of creating a single `event` with all records, we could create multiple events according to the batch size configuration of the event source mapping here:

https://github.com/localstack/localstack/blob/master/localstack/services/awslambda/lambda_api.py#L210-L219

Would you be able to create a pull request and contribute a simple integration test? That would be great! Thanks

Commit message: Fix issue #1131
* Fix long-line issue with lint

Patch to review:
diff --git a/localstack/services/awslambda/lambda_api.py b/localstack/services/awslambda/lambda_api.py
index 1d207e6556b66..7a705516b3c4e 100644
--- a/localstack/services/awslambda/lambda_api.py
+++ b/localstack/services/awslambda/lambda_api.py
@@ -152,12 +152,12 @@ def add_function_mapping(lambda_name, lambda_handler, lambda_cwd=None):
     arn_to_lambda[arn].cwd = lambda_cwd
 
 
-def add_event_source(function_name, source_arn, enabled):
+def add_event_source(function_name, source_arn, enabled, batch_size=10):
     mapping = {
         'UUID': str(uuid.uuid4()),
         'StateTransitionReason': 'User action',
         'LastModified': float(time.mktime(datetime.utcnow().timetuple())),
-        'BatchSize': 100,
+        'BatchSize': batch_size,
         'State': 'Enabled' if enabled is True or enabled is None else 'Disabled',
         'FunctionArn': func_arn(function_name),
         'EventSourceArn': source_arn,
@@ -245,22 +245,31 @@ def process_sns_notification(func_arn, topic_arn, subscriptionArn, message, mess
 
 
 def process_kinesis_records(records, stream_name):
+    def chunks(lst, n):
+        # Yield successive n-sized chunks from lst.
+        for i in range(0, len(lst), n):
+            yield lst[i:i + n]
+
     # feed records into listening lambdas
     try:
         stream_arn = aws_stack.kinesis_stream_arn(stream_name)
         sources = get_event_sources(source_arn=stream_arn)
         for source in sources:
             arn = source['FunctionArn']
-            event = {
-                'Records': []
-            }
-            for rec in records:
-                event['Records'].append({
-                    'eventID': 'shardId-000000000000:{0}'.format(rec['sequenceNumber']),
-                    'eventSourceARN': stream_arn,
-                    'kinesis': rec
-                })
-            run_lambda(event=event, context={}, func_arn=arn)
+            for chunk in chunks(records, source['BatchSize']):
+                event = {
+                    'Records': [
+                        {
+                            'eventID': 'shardId-000000000000:{0}'.format(rec['sequenceNumber']),
+                            'eventSourceARN': stream_arn,
+                            'kinesis': rec
+                        }
+                        for rec in chunk
+                    ]
+                }
+
+                run_lambda(event=event, context={}, func_arn=arn)
+
     except Exception as e:
         LOG.warning('Unable to run Lambda function on Kinesis records: %s %s' % (e, traceback.format_exc()))
 
@@ -1134,7 +1143,9 @@ def create_event_source_mapping():
               in: body
     """
     data = json.loads(to_str(request.data))
-    mapping = add_event_source(data['FunctionName'], data['EventSourceArn'], data.get('Enabled'))
+    mapping = add_event_source(
+        data['FunctionName'], data['EventSourceArn'], data.get('Enabled'), data.get('BatchSize', 10)
+    )
     return jsonify(mapping)
 
 


Code review comments:
diff_hunk: @@ -1134,7 +1143,9 @@ def create_event_source_mapping():
               in: body
     """
     data = json.loads(to_str(request.data))
-    mapping = add_event_source(data['FunctionName'], data['EventSourceArn'], data.get('Enabled'))
+    mapping = add_event_source(
+        data['FunctionName'], data['EventSourceArn'], data.get('Enabled'), data.get('BatchSize', 10)
path: localstack/services/awslambda/lambda_api.py
text: @user1:
nit: perhaps we can remove the default value (`10`) here and handle that case in `add_event_source` (see comment above).
```

    mapping = add_event_source(
        data['FunctionName'], data['EventSourceArn'], data.get('Enabled'), data.get('BatchSize') 
    )
```

diff_hunk: @@ -152,12 +152,12 @@ def add_function_mapping(lambda_name, lambda_handler, lambda_cwd=None):
     arn_to_lambda[arn].cwd = lambda_cwd
 
 
-def add_event_source(function_name, source_arn, enabled):
+def add_event_source(function_name, source_arn, enabled, batch_size=10):
path: localstack/services/awslambda/lambda_api.py
text: @user1:
Instead of encoding the default value in the function signature, I'd rather handle the default case in the function body (we can define `DEFAULT_BATCH_SIZE` towards top of the file):
```
DEFAULT_BATCH_SIZE = 10

...

def add_event_source(function_name, source_arn, enabled, batch_size=None): 
    batch_size = batch_size or DEFAULT_BATCH_SIZE
    ...
```