add: Support for ESM v2 partial batch failure handling (Kinesis & DynamoDB) #9
Conversation
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
PR Summary
This PR adds support for partial batch failure handling in Event Source Mapping (ESM) v2 for Kinesis, DynamoDB, and SQS in LocalStack's Lambda service.
- Implemented
FunctionResponseTypessupport inEsmConfigFactoryfor SQS, Kinesis, and DynamoDB Streams - Enhanced
StreamPollerto handlePartialBatchFailureError, including retry logic for failed items - Added
report_batch_item_failuresflag inLambdaSenderto enable partial success/failure processing - Introduced new test cases in
test_lambda_integration_dynamodbstreams.pyandtest_lambda_integration_kinesis.pyfor partial batch failure scenarios - Updated
utils.pywith new Lambda function templates for DynamoDB and Kinesis batch item failure handling
16 file(s) reviewed, 9 comment(s)
Edit PR Review Bot Settings
| ) | ||
|
|
||
| # TODO What happens when FunctionResponseTypes value or target service is invalid? | ||
| if service in ["sqs", "kinesis", "dynamodbstreams"]: |
There was a problem hiding this comment.
style: Use a constant or enum for the list of services that support FunctionResponseTypes
| def bisect_events( | ||
| self, sequence_number: str, events: list[dict] | ||
| ) -> tuple[list[dict], list[dict]]: | ||
| for i, event in enumerate(events): | ||
| if self.get_sequence_number(event) == sequence_number: | ||
| return events[:i], events[i:] | ||
|
|
||
| return events, [] |
There was a problem hiding this comment.
style: Consider optimizing bisect_events for large event lists, potentially using binary search
| @@ -19,9 +19,20 @@ class LambdaSender(Sender): | |||
| # Flag to enable the payload dict using the "Records" key used for Lambda event source mapping | |||
| payload_dict: bool | |||
|
|
|||
| def __init__(self, target_arn, target_parameters=None, target_client=None, payload_dict=False): | |||
| # Flag to enable partial successes/failures when processing batched events through a Lambda event source mapping | |||
| report_batch_item_failures: bool | |||
There was a problem hiding this comment.
style: Consider adding type hints for boolean flags
| expected_successes = 5 | ||
| expected_failures = 1 |
There was a problem hiding this comment.
style: Consider using a constant or configuration value for these numbers
| @@ -578,7 +563,7 @@ def test_report_batch_item_failures( | |||
| ) | |||
| snapshot.match("first_invocation", first_invocation) | |||
|
|
|||
| # check that the DQL is empty | |||
| # check that the DLQ is empty | |||
| dlq_messages = aws_client.sqs.receive_message(QueueUrl=event_dlq_url) | |||
| @@ -578,7 +563,7 @@ def test_report_batch_item_failures( | |||
| ) | |||
| snapshot.match("first_invocation", first_invocation) | |||
|
|
|||
| # check that the DQL is empty | |||
| # check that the DLQ is empty | |||
| dlq_messages = aws_client.sqs.receive_message(QueueUrl=event_dlq_url) | |||
There was a problem hiding this comment.
syntax: DLQ misspelled as DQL
| dlq_messages = aws_client.sqs.receive_message(QueueUrl=event_dlq_url) | |
| # check that the DLQ is empty |
| if new_image.get("should_fail", {}).get("BOOL", False): | ||
| batch_item_failures.append({"itemIdentifier": record["dynamodb"]["SequenceNumber"]}) |
There was a problem hiding this comment.
style: Consider using .get('BOOL') instead of .get('BOOL', False) for consistency with the Kinesis handler
| if payload.get("should_fail", False): | ||
| batch_item_failures.append({"itemIdentifier": record["kinesis"]["sequenceNumber"]}) | ||
|
|
||
| return {"batchItemFailures" : batch_item_failures} |
There was a problem hiding this comment.
syntax: Extra space before the colon in 'batchItemFailures :'
| def create_lambda_with_response(response: str) -> str: | ||
| """Creates a lambda with pre-defined response""" | ||
| return _LAMBDA_WITH_RESPONSE.format(response=response) |
There was a problem hiding this comment.
style: Add type hinting for the 'response' parameter
Motivation
This PR adds support for
FunctionResponseTypesto ESM v2, allowing for partial batch failures to be reported and handled. Failed items of a batch can now be retried in accordance with aMaximumRetryAttemptspolicy.Changes
Event Source Mapping: SQS
Event Source Mapping: Kinesis & DynamoDB
Testing
Partial Failure with ReportBatchItemFailures
Both tests simulates a partial batch failure, capturing failure information in an OnFailure destination config using a DLQ:
Success and failure conditions
All success and failure cases outlined in the ESM docs have test coverage (where conditions are identical for all Kinesis, DynamoDB, and SQS). These are covered in the following parametrized tests:
Total Batch Successes:
test_kinesis_report_batch_item_success_scenariostest_dynamodb_report_batch_item_success_scenariosTotal Batch Failures:
test_kinesis_report_batch_item_failure_scenariostest_dynamodb_report_batch_item_failure_scenarios