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

Aberta
  • Algo errado?
#8,661 2 comentários 0 reações 1 responsável Ver no GitHub

Mantenedores costumam responder em até 2 dias

@soujasK já está trabalhando nisso.

Desde 25/10/2026.

  • #8663 de @soujasK — aberto

Avaliação

Esta issue ainda não foi avaliada.

Descrição

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

Linguagem predominante
Python
Estrelas
3.3k
Forks
511
Merge médio
3d 12h
PRs com merge (30d)
93

Preparar o ambiente

Primeiros passos

  1. Leia a issue inteira e depois o guia de contribuição do projeto.
  2. Comente na issue dizendo que vai assumir — evita que duas pessoas façam o mesmo trabalho.
  3. Faça um fork do repositório e trabalhe em uma branch.
  4. Abra um pull request que referencie o número da issue.

Mais de aws-powertools/powertools-lambda-python

Todas as issues de aws-powertools/powertools-lambda-python

Issues semelhantes

Mais issues de Python

Receba novas issues na sua caixa de entrada

Um resumo curto de issues do GitHub para quem está começando.