RFC: New batch processor for new native partial response (SQS, DynamoDB, Kinesis)

[to-mc]

Key information

• RFC PR: feat(batch): new BatchProcessor for SQS, DynamoDB, Kinesis powertools-lambda-python#886
• Related issue(s), if known:
• Area: Batch processing
• Meet tenets: Yes
• Approved by: @heitorlessa
• Reviewed by: @heitorlessa

Summary

A new generic batch processing utility, which can process records from SQS, Kinesis Data Streams, and DynamoDB streams, and handle reporting batch failures.

Motivation

With the launch of support for partial batch responses for Lambda/SQS, the event source mapping can now natively handle partial failures in a batch - removing the need for calls to the delete api. This support already exists for Kinesis and DynamoDB streams.

Proposal

The new utility will be implemented in the namespace which already exists for the SQS Batch Processing utility - aws_lambda_powertools.utilities.batch. Rather than calling the SQS DeleteMessage api like the existing batch utility, this version will instead inject BatchItemFailures (or batchItemFailures for kinesis and ddb) into the Lambda response. We will expose at the least, a new decorator batch_processor, which will accept an event type depending on the integration (Kinesis, SQS or DDB Streams). This will look similar to the Event Handler design. There will be a boolean setting to handle partial failures, defaulting to True (users will still need to enable this in the event source mapping for it to work).

from aws_lambda_powertools.batch import KinesisStreamEvent, EventType


def record_handler(record):
    return do_something_with(record["body"])


@batch_processor(record_handler=record_handler, event_type=EventType.KinesisStreamEvent)
def lambda_handler(event, context):
    return {"statusCode": 200} 

Proposed class/method names for the new api:
aws_lambda_powertools.utilities.batch.base.BatchProcessor
aws_lambda_powertools.utilities.batch.batch_processor

Any configuration or corner cases you'd expect?
Users will need to enable this in the event source mapping when configuring the Lambda trigger - reporting partial failures will not work with only changes to the Lambda code. Investigation is needed to better understand the consequences here, and how expensive it would be to enable a check in the code to see if it is enabled or not. If we don't implement this, we need to call this out front and center in the documentation.

Rationale and alternatives

The main consideration here is where the new functionality fits into the powertools package. It could be a new top level utility eg. aws_lambda_powertools.batchv2 - but we prefer not to add version numbers as it is confusing for users.

We could make a straight replacement of the implementation behind the existing API, which was the initial idea. However, the native functionality requires a setting when the event source mapping is created. That means we'd be introducing a breaking change if we did this - the utility would stop working for users who had not made this configuration.

Unresolved questions

Need to decide if we should add a context manager like the existing implementation has. It is simple to implement this, but it burdens the user with ensuring they store the return value and use it in their lambda response. I feel like this is too likely to be misunderstood, but would like opinions. Example:

from aws_lambda_powertools.utilities.batch import BatchProcessor

def record_handler(record):
    return do_something_with(record["body"])

def lambda_handler(event, context):
    records = event["Records"]

    processor = BatchProcessor(event_type=EventType.KinesisStreamEvent)

    with processor(records, record_handler) as proc:
        result = proc.process()  # Users will have to store the response from processor
  
   return result. # Users will have to return the result they stored from the processor

[kimberlyamandalu]

I think the batch_processor decorator name might already be taken by the following: Create your own Partial Processor. If I'm reading this incorrectly, I apologize and pls correct me.

Does it make sense to potentially leverage on that capability to implement the "new" batch processors?

[heitorlessa]

You’re right Kim, missed that somehow when looking at top imports and search. As it expects a processor, there might be a way to make this non-breaking change — make a default new Processor, and introduce an optional BatchType argument. If we can find a better name would be a better solution tho https://github.com/awslabs/aws-lambda-powertools-python/blob/develop/aws_lambda_powertools/utilities/batch/base.py

…

On Tue, 30 Nov 2021 at 21:59, kimberlyamandalu ***@***.***> wrote: I think the batch_processor decorator name might already be taken by the following: Create your own Partial Processor <https://awslabs.github.io/aws-lambda-powertools-python/latest/utilities/batch/#create-your-own-partial-processor>. If I'm reading this incorrectly, I apologize and pls correct me. Does it make sense to potentially leverage on that capability to implement the "new" batch processors? — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <https://github.com/awslabs/aws-lambda-powertools-python/issues/858#issuecomment-983015247>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAZPQBH5FAKZDILKGA3TOMDUOU3LTANCNFSM5JBLFF5A> .

[heitorlessa]

What about @report_batch_failures(batch_type=, record_handler=)?

That is explicit enough and should help customers find it more easily if they've read the news, found somewhere and came looking for support.

[kimberlyamandalu]

I think that is a pretty good name for it since sqs/dynamodb/kinesis all refer to it as such.
The only thing I could think of that might be confusing is, it might make it look like it is just an error handling feature. Which I guess it kind of is, but more :)

I honestly cannot think of a better name, other than @partial_batch_processor maybe?

But I'm sure whatever becomes its name, it will be described clearly in the document and searches on google should return relevant results.

[gmcrocetti]

Hi folks, long time no see :).
I'm really thrilled seeing AWS finally provided such feature for SQS 🚀 .

Why not leverage current @batch_processor instead of creating a new decorator with "similar" behavior. Can't we achieve the same result by doing the following:

1. Adding a inject_batchitem_failures as an argument to @batch_processor;
2. Writing a new PartialProcessor extending BasePartialProcessor - not 100% sure we need it.

def batchitem_failure_handler(event_source: str, failed_records: List[Dict]) -> Dict:
    key = "batchItemFailures" if event_source in ("aws:kinesis", "aws:dynamodb") else "BatchItemFailures"
    failed_batch_items = [{"ItemIdentifier": failed_record["eventID"]} for failed_record in failed_records]
    return {key: failed_batch_items}


@lambda_handler_decorator
def batch_processor(
    handler: Callable, 
    event: Dict, 
    context: Dict, 
    record_handler: Callable, 
    processor: BasePartialProcessor, 
    inject_batchitem_failures: bool = False,  # non-breaking change
):
    records = event["Records"]
    # Do we really need `event_type` as an argument ?
    event_source = event["eventSource"]

    with processor(records, record_handler):
        processor.process()

    result = handler(event, context)
    
    if inject_batchitem_failures:
        failure_response = batchitem_failure_handler(event_source, processor.fail_messages)
        result.update(failure_response)
    
    return result

I'm in a rush so sorry if I missed a critical point. Looking forward to hear from you 👍

[heitorlessa]

Yes, I mentioned that to @cakepietoast but he had a pretty good argument on the contrary -- can't remember what it was tho :/

[ran-isenberg]

@heitorlessa other than automatic handling of error, this can already be achieved with the current Parser utility.
why not extend the parser instead of creating a new utility?

We process SQS/dynamo streams. batches with the parser and it works great.

[heitorlessa]

Ran, there’s one more aspect that we only implemented in Java - Move non-retryable exceptions to the DLQ. We wouldn’t want to make API calls underneath when using Parser. That said, we could have a better integration with Parser but we haven’t thought how Parser and the lightweight Data classes could coexist here yet - it needs a RFC/ design session

…

On Sun, 5 Dec 2021 at 07:28, Ran Isenberg ***@***.***> wrote: @heitorlessa <https://github.com/heitorlessa> other than automatic handling of error, this can already be achieved with the current Parser utility. why not extend the parser instead of creating a new utility? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <https://github.com/awslabs/aws-lambda-powertools-python/issues/858#issuecomment-986174977>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAZPQBB3PRV7H7GLPTD53DTUPMBATANCNFSM5JBLFF5A> .

[heitorlessa]

We've made a decision - we'll create a new Processor that handles all 3 (SQS, DynamoDB Stream, Kinesis Data Streams) and have no additional arguments for the batch_processor to make it easier to get it right + deprecating things in v2 easier.

Gonna start working on the implementation now, and we should have it ready for the next release early next week

[ran-isenberg]

@heitorlessa Do you think there should be a Parser equivalent? I think it can easily to integrate the processing or failed processing that you write here and add it as an extra parameter to the parser.

[heitorlessa]

Could you give an example @ran-isenberg ? I can see how we could have an extra parameter to receive a Pydantic model OR an event source data class (SQS), so the record handler receives a strongly typed record.

---

Everyone, this is how the initial UX is looking like now - We'll work on Kinesis/DynamoDB next, but the experience will be a simple toggle e.g., EventType.DynamoDB, everything else remains the same.

As a decorator

from aws_lambda_powertools.utilities.batch import batch_processor, BatchProcessor, EventType

processor = BatchProcessor(event_type=EventType.SQS)


@batch_processor(record_handler=record_handler, processor=processor)
def lambda_handler(event, context):
    return processor.report() # this is necessary to ensure Lambda returns the correct response

As a context manager, in case you need full access to the batch processed, or handle exceptions yourself

from aws_lambda_powertools.utilities.batch import batch_processor, BatchProcessor, EventType

processor = BatchProcessor(event_type=EventType.SQS)


def lambda_handler(event, context):
    with processor(records, record_handler) as batch:
        processed_messages = batch.process()
    
    return batch.report() # this is necessary to ensure Lambda returns the correct response

[michaelbrewer]

@heitorlessa fitting that into the lightweight event source classes would be neat

[gmcrocetti]

We've made a decision - we'll create a new Processor that handles all 3 (SQS, DynamoDB Stream, Kinesis Data Streams) and have no additional arguments for the batch_processor to make it easier to get it right + deprecating things in v2 easier.

Gonna start working on the implementation now, and we should have it ready for the next release early next week

Really liked the decision you guys made. Writing a new processor that injects failed messages is the best of both worlds. 🚀
Just to give another 👍, adding up event source classes to this new processor will be super neat. Can't wait to see the RFC ! :) .

Just a small nit. I can't see to find BatchItemFailures - with capital B. Looks like someone corrected the typo on the documentation ? 🤔

[ran-isenberg]

@heitorlessa I think it makes a lot of sense to integrate this utility with any of the 2 input validation tools- parser/event source.
Maybe having 2 decorators -one in the extras package for the parser and one in the regular.
In the end, I think both decorators would be able to call a shared code part once they finish handling the event with either the parser/event source utilities.

I wrote something like that with the parser for SQS and dynamodb streams which iterates messages, catches processing exceptions, adds failed sqs messages to a list , continues the loop and in the end sends failed messages to a DLQ. So I know it can be done :)

[heitorlessa]

We've made a decision - we'll create a new Processor that handles all 3 (SQS, DynamoDB Stream, Kinesis Data Streams) and have no additional arguments for the batch_processor to make it easier to get it right + deprecating things in v2 easier.
Gonna start working on the implementation now, and we should have it ready for the next release early next week

Really liked the decision you guys made. Writing a new processor that injects failed messages is the best of both worlds. 🚀 Just to give another 👍, adding up event source classes to this new processor will be super neat. Can't wait to see the RFC ! :) .

Just a small nit. I can't see to find BatchItemFailures - with capital B. Looks like someone corrected the typo on the documentation ? 🤔

They've corrected in the docs. The docs that went live as part of the launch had a capital B, they have been corrected after this RFC was created ;)