From 1545ba02b28f3a080dc5806eb3d050871b797677 Mon Sep 17 00:00:00 2001 From: Soujas K Date: Sun, 25 Oct 2026 21:55:32 +0530 Subject: [PATCH] fix(batch): report short-circuited FIFO records that don't match the model --- .../batch/sqs_fifo_partial_processor.py | 8 +++++- .../test_utilities_batch_pydantic.py | 28 +++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/aws_lambda_powertools/utilities/batch/sqs_fifo_partial_processor.py b/aws_lambda_powertools/utilities/batch/sqs_fifo_partial_processor.py index 88a6c3dfe37..2f8d9b4a4ad 100644 --- a/aws_lambda_powertools/utilities/batch/sqs_fifo_partial_processor.py +++ b/aws_lambda_powertools/utilities/batch/sqs_fifo_partial_processor.py @@ -100,8 +100,14 @@ def _process_record(self, record): fail_entire_batch = bool(self.fail_messages) and not self._skip_group_on_error fail_group_id = self._skip_group_on_error and self._current_group_id in self._failed_group_ids if fail_entire_batch or fail_group_id: + try: + record_to_fail = self._to_batch_type(record, event_type=self.event_type, model=self.model) + except Exception: + # The record never reaches the handler, so register it as a data class like other poison pills + record_to_fail = self._to_batch_type(record, event_type=self.event_type) + return self.failure_handler( - record=self._to_batch_type(record, event_type=self.event_type, model=self.model), + record=record_to_fail, exception=self.group_circuit_breaker_exc if self._skip_group_on_error else self.circuit_breaker_exc, ) diff --git a/tests/functional/batch/_pydantic/test_utilities_batch_pydantic.py b/tests/functional/batch/_pydantic/test_utilities_batch_pydantic.py index 15a7c681fbf..2ccabe1d56d 100644 --- a/tests/functional/batch/_pydantic/test_utilities_batch_pydantic.py +++ b/tests/functional/batch/_pydantic/test_utilities_batch_pydantic.py @@ -486,6 +486,34 @@ def lambda_handler(event, context): assert result["batchItemFailures"][2]["itemIdentifier"] == fifth_record.message_id +@pytest.mark.parametrize("skip_group_on_error", [False, True]) +def test_sqs_fifo_batch_processor_model_with_validation_error_after_failure( + skip_group_on_error: bool, + record_handler_model: Callable, + sqs_event_fifo_factory, + order_event_factory, +): + # GIVEN a malformed record that is short-circuited because a previous record of its group failed + first_record = sqs_event_fifo_factory(order_event_factory({"type": "success"}), "2") + failed_record = sqs_event_fifo_factory(order_event_factory({"type": "fail"}), "1") + malformed_record = sqs_event_fifo_factory(json.dumps({"poison": "pill"}), "1") + fourth_record = sqs_event_fifo_factory(order_event_factory({"type": "success"}), "2") + records = [first_record, failed_record, malformed_record, fourth_record] + + # WHEN + processor = SqsFifoPartialProcessor(model=OrderSqs, skip_group_on_error=skip_group_on_error) + with processor(records, record_handler_model) as batch: + batch.process() + + # THEN the malformed record is reported instead of failing the whole batch + expected_failures = [failed_record, malformed_record] + if not skip_group_on_error: + expected_failures.append(fourth_record) + assert batch.response() == { + "batchItemFailures": [{"itemIdentifier": record["messageId"]} for record in expected_failures], + } + + def test_batch_processor_model_with_partial_validation_error( record_handler_model: Callable, sqs_event_factory,