add: Support for ESM v2 partial batch failure handling (Kinesis & DynamoDB)

localstack-core/localstack/services/lambda_/event_source_mapping/esm_config_factory.py

@@ -73,9 +73,12 @@ def get_esm_config(self) -> EventSourceMappingConfiguration:
         # c) Are FilterCriteria.Filters merged or replaced upon update?
         # TODO: can we ignore extra parameters from the request (e.g., Kinesis params for SQS source)?
         derived_source_parameters = merge_recursive(default_source_parameters, self.request)
-        derived_source_parameters["FunctionResponseTypes"] = derived_source_parameters.get(
-            "FunctionResponseTypes", []
-        )
+
+        # TODO What happens when FunctionResponseTypes value or target service is invalid?
+        if service in ["sqs", "kinesis", "dynamodbstreams"]:
+            derived_source_parameters["FunctionResponseTypes"] = derived_source_parameters.get(
+                "FunctionResponseTypes", []
+            )
 
         state = EsmState.CREATING if self.request.get("Enabled", True) else EsmState.DISABLED
         esm_config = EventSourceMappingConfiguration(

localstack-core/localstack/services/lambda_/event_source_mapping/esm_event_processor.py

@@ -58,7 +58,9 @@ def process_events_batch(self, input_events: list[dict]) -> None:
             # TODO: check whether partial batch item failures is enabled by default or need to be explicitly enabled
             #  using --function-response-types "ReportBatchItemFailures"
             #  https://docs.aws.amazon.com/lambda/latest/dg/services-sqs-errorhandling.html
-            raise PartialBatchFailureError from e
+            raise PartialBatchFailureError(
+                partial_failure_payload=e.partial_failure_payload, error=e.error
+            ) from e
         except SenderError as e:
             self.logger.log(
                 messageType="ExecutionFailed",

localstack-core/localstack/services/lambda_/event_source_mapping/esm_worker_factory.py

@@ -1,5 +1,6 @@
 from localstack.aws.api.lambda_ import (
     EventSourceMappingConfiguration,
+    FunctionResponseType,
 )
 from localstack.aws.api.pipes import (
     DynamoDBStreamStartPosition,
@@ -59,6 +60,8 @@ def get_esm_worker(self) -> EsmWorker:
             ),
             target_client=lambda_client,
             payload_dict=True,
+            report_batch_item_failures=self.esm_config.get("FunctionResponseTypes")
+            == [FunctionResponseType.ReportBatchItemFailures],
         )
 
         # Logger

localstack-core/localstack/services/lambda_/event_source_mapping/event_processor.py

@@ -46,7 +46,9 @@ class PartialBatchFailureError(EventProcessorError):
     def __init__(
         self,
         partial_failure_payload: PartialFailurePayload | None = None,
+        error=None,
     ) -> None:
+        self.error = error
         self.partial_failure_payload = partial_failure_payload
 
 

localstack-core/localstack/services/lambda_/event_source_mapping/pollers/poller.py

@@ -121,10 +121,25 @@ def has_batch_item_failures(
     try:
         failed_items_ids = parse_batch_item_failures(result, valid_item_ids)
         return len(failed_items_ids) > 0
-    except KeyError:
+    except (KeyError, ValueError):
         return True
 
 
+def get_batch_item_failures(
+    result: dict | str | None, valid_item_ids: set[str] | None = None
+) -> list[str] | None:
+    """
+    Returns a list of failed batch item IDs. If an empty list is returned, then the batch should be considered as a complete success.
+
+    If `None` is returned, the batch should be considered a complete failure.
+    """
+    try:
+        failed_items_ids = parse_batch_item_failures(result, valid_item_ids)
+        return failed_items_ids
+    except (KeyError, ValueError):
+        return None
+
+
 def parse_batch_item_failures(
     result: dict | str | None, valid_item_ids: set[str] | None = None
 ) -> list[str]:
@@ -178,6 +193,8 @@ def parse_batch_item_failures(
             raise KeyError(f"missing itemIdentifier in batchItemFailure record {item}")
 
         item_identifier = item["itemIdentifier"]
+        if not item_identifier:
+            raise ValueError("itemIdentifier cannot be empty or null")
 
         # Optionally validate whether the item_identifier is part of the batch
         if valid_item_ids and item_identifier not in valid_item_ids:

localstack-core/localstack/services/lambda_/event_source_mapping/pollers/stream_poller.py

@@ -22,12 +22,11 @@
     get_datetime_from_timestamp,
     get_internal_client,
 )
-from localstack.services.lambda_.event_source_mapping.pollers.poller import Poller
-from localstack.services.lambda_.event_source_mapping.pollers.sqs_poller import get_queue_url
-from localstack.services.lambda_.event_source_mapping.senders.sender import (
-    PartialFailureSenderError,
-    SenderError,
+from localstack.services.lambda_.event_source_mapping.pollers.poller import (
+    Poller,
+    get_batch_item_failures,
 )
+from localstack.services.lambda_.event_source_mapping.pollers.sqs_poller import get_queue_url
 from localstack.utils.aws.arns import parse_arn
 
 LOG = logging.getLogger(__name__)
@@ -192,18 +191,40 @@ def poll_events_from_shard(self, shard_id: str, shard_iterator: str):
             except PartialBatchFailureError as ex:
                 # TODO: add tests for partial batch failure scenarios
                 if (
-                    self.stream_parameters["OnPartialBatchItemFailure"]
+                    self.stream_parameters.get("OnPartialBatchItemFailure")
                     == OnPartialBatchItemFailureStreams.AUTOMATIC_BISECT
                 ):
                     # TODO: implement and test splitting batches in half until batch size 1
                     #  https://docs.aws.amazon.com/eventbridge/latest/pipes-reference/API_PipeSourceKinesisStreamParameters.html
                     LOG.warning(
                         "AUTOMATIC_BISECT upon partial batch item failure is not yet implemented. Retrying the entire batch."
                     )
-                error_payload = ex.partial_failure_payload
-                # let entire batch fail (ideally raise BatchFailureError)
-            except (SenderError, PartialFailureSenderError, BatchFailureError, Exception) as ex:
-                if isinstance(ex, (SenderError, PartialFailureSenderError, BatchFailureError)):
+                error_payload = ex.error
+
+                # If the batchItemFailures array contains multiple items, Lambda uses the record with the lowest sequence number as the checkpoint.
+                # Lambda then retries all records starting from that checkpoint.
+
+                failed_sequence_ids: list[int] | None = get_batch_item_failures(
+                    ex.partial_failure_payload
+                )
+
+                # If None is returned, consider the entire batch a failure.
+                if failed_sequence_ids is None:
+                    continue
+
+                # This shouldn't be possible since a PartialBatchFailureError was raised
+                if len(failed_sequence_ids) == 0:
+                    LOG.warning(
+                        "Invalid state encountered: PartialBatchFailureError raised but no batch item failures found."
+                    )
+                    return
+
+                lowest_sequence_id: str = min(failed_sequence_ids, key=int)
+
+                # Discard all successful events and re-process from sequence number of failed event
+                _, events = self.bisect_events(lowest_sequence_id, events)
+            except (BatchFailureError, Exception) as ex:
+                if isinstance(ex, BatchFailureError):
                     error_payload = ex.error
 
                 # FIXME partner_resource_arn is not defined in ESM
@@ -213,11 +234,9 @@ def poll_events_from_shard(self, shard_id: str, shard_iterator: str):
                     self.partner_resource_arn or self.source_arn,
                     events,
                 )
-                attempts += 1
+            finally:
                 # Retry polling until the record expires at the source
-                if self.stream_parameters.get("MaximumRetryAttempts", -1) == -1:
-                    # TODO: handle iterator expired scenario
-                    return
+                attempts += 1
 
         # Send failed events to potential DLQ
         abort_condition = abort_condition or "RetryAttemptsExhausted"
@@ -324,3 +343,12 @@ def max_retries_exceeded(self, attempts: int) -> bool:
         if maximum_retry_attempts == -1:
             return False
         return attempts > maximum_retry_attempts
+
+    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, []

localstack-core/localstack/services/lambda_/event_source_mapping/senders/lambda_sender.py

@@ -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
+
+    def __init__(
+        self,
+        target_arn,
+        target_parameters=None,
+        target_client=None,
+        payload_dict=False,
+        report_batch_item_failures=False,
+    ):
         super().__init__(target_arn, target_parameters, target_client)
         self.payload_dict = payload_dict
+        self.report_batch_item_failures = report_batch_item_failures
 
     def send_events(self, events: list[dict]) -> dict:
         if self.payload_dict:
@@ -72,18 +83,17 @@ def send_events(self, events: list[dict]) -> dict:
                 error=error,
             )
 
-        # TODO: test all success, partial, and failure conditions:
-        #   https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-dynamodb.html#pipes-ddb-batch-failures
         # The payload can contain the key "batchItemFailures" with a list of partial batch failures:
         # https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-batching-concurrency.html
-        if has_batch_item_failures(payload):
+        if self.report_batch_item_failures and has_batch_item_failures(payload):
             error = {
                 "message": "Target invocation failed partially.",
                 "httpStatusCode": invoke_result["StatusCode"],
                 "awsService": "lambda",
                 "requestId": invoke_result["ResponseMetadata"]["RequestId"],
                 "exceptionType": "BadRequest",
                 "resourceArn": self.target_arn,
+                "executedVersion": invoke_result.get("ExecutedVersion", "$LATEST"),
             }
             raise PartialFailureSenderError(error=error, partial_failure_payload=payload)
 

localstack-core/localstack/services/lambda_/event_source_mapping/senders/sender.py

@@ -7,7 +7,7 @@
 
 class SenderError(Exception):
     def __init__(self, message=None, error=None) -> None:
-        self.message = message or "Error during during sending events"
+        self.message = message or "Error during sending events"
         self.error = error