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

Aperta
  • Qualcosa non va?
#8,661 2 commenti 0 reazioni 1 assegnatario Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@soujasK ci sta già lavorando.

Dal 25/10/2026.

  • #8663 di @soujasK — aperta

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

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

Lingua principale
Python
Stelle
3.3k
Fork
511
Merge medio
3g 8h
PR unite (30g)
92

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di aws-powertools/powertools-lambda-python

Tutte le issue di aws-powertools/powertools-lambda-python

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.