Bug: SqsFifoPartialProcessor fails the whole batch when a short-circuited record doesn't match the model

Offen
  • Stimmt etwas nicht?
#8,661 2 Kommentare 0 Reaktionen 1 zugewiesene Person Auf GitHub ansehen

Maintainer antworten meist innerhalb von 2 Tagen

@soujasK arbeitet bereits daran.

Seit 25.10.2026.

  • #8663 von @soujasK — offen

Bewertung

Dieses Issue wurde noch nicht bewertet.

Beschreibung

batch bug good first issue
Expected Behaviour

With SqsFifoPartialProcessor(model=...), once a record fails, the remaining records (or the remaining records of the failed message group with skip_group_on_error=True) should be reported in batchItemFailures, even if one of them doesn't match the model.

Current Behaviour

The short-circuit path converts the record with the customer's model outside of any try:

if fail_entire_batch or fail_group_id:
    return self.failure_handler(
        record=self._to_batch_type(record, event_type=self.event_type, model=self.model),
        exception=...,
    )

If that record fails validation, the ValidationError propagates out of process() and the whole invocation fails. The poison-pill handling from #2091 never runs here because it lives in BatchProcessor._process_record, which isn't called for short-circuited records.

As a result the partial batch response is lost and every record is retried, including the ones that were already processed successfully. With skip_group_on_error=True that includes the successful records from other message groups.

(Different code path from #8660, which is about BatchProcessor._process_record.)

Code snippet
from pydantic import BaseModel, Json

from aws_lambda_powertools.utilities.batch import SqsFifoPartialProcessor, process_partial_response
from aws_lambda_powertools.utilities.parser.models import SqsRecordModel


class OrderBody(BaseModel):
    order_id: int


class Order(SqsRecordModel):
    body: Json[OrderBody]


def make_record(message_id, body, group_id):
    return {
        "messageId": message_id,
        "receiptHandle": "handle",
        "body": body,
        "attributes": {
            "ApproximateReceiveCount": "1",
            "SentTimestamp": "1",
            "SenderId": "sender",
            "ApproximateFirstReceiveTimestamp": "1",
            "MessageGroupId": group_id,
            "SequenceNumber": "1",
            "MessageDeduplicationId": "dedup",
        },
        "messageAttributes": {},
        "md5OfBody": "md5",
        "eventSource": "aws:sqs",
        "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:queue.fifo",
        "awsRegion": "us-east-1",
    }


def record_handler(record: Order):
    if record.body.order_id == 1:
        raise RuntimeError("failed to process order")


event = {
    "Records": [
        make_record("1", '{"order_id": 1}', "group-1"),  # fails in the handler
        make_record("2", '{"unexpected": 1}', "group-1"),  # doesn't match the model
        make_record("3", '{"order_id": 3}', "group-2"),
    ],
}

for skip_group_on_error in (False, True):
    processor = SqsFifoPartialProcessor(model=Order, skip_group_on_error=skip_group_on_error)
    # Expected: batchItemFailures for 1 and 2 (and 3 when skip_group_on_error=False)
    # Actual: pydantic ValidationError for record 2 escapes and fails the invocation
    print(process_partial_response(event=event, record_handler=record_handler, processor=processor))
Possible Solution

The short-circuited record is never passed to the record handler, so it only needs to be convertible for failure_handler / _collect_sqs_failures, which already accept event source data classes for poison pills. Two options:

  1. Fall back to the event source data class when the model conversion fails (keeps fail_messages as model instances for valid records, same as today).
  2. Always register short-circuited records as event source data classes. Simpler, but changes the type of entries in fail_messages for customers who inspect them.

I'd go with 1 and add tests for both skip_group_on_error modes. Happy to open a PR.

Steps to Reproduce

Run the snippet above.

Powertools for AWS Lambda (Python) version

latest (develop @ 45f7a4d)

AWS Lambda function runtime

3.11

Packaging format used

PyPi

Debugging logs

No response

Vorherrschende Sprache
Python
Sterne
3.3k
Forks
511
Ø Merge
3 T. 12 Std.
Gemergte PRs (30 T.)
93

Entwicklungsumgebung

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus aws-powertools/powertools-lambda-python

Alle Issues in aws-powertools/powertools-lambda-python

Ähnliche Issues

Weitere Issues zu Python

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.