Idempotency config warning - couldn't determine time left

[andrewhharmon]

I'm trying to process some SQS records using batch processing and idempotency. I assumed I needed to decorate my business logic function so each record is processed only once. Following the sample here. I set my code up as follows.

from aws_lambda_powertools.utilities.idempotency import (
    DynamoDBPersistenceLayer, IdempotencyConfig, idempotent_function)
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, batch_processor
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord

from aws_lambda_powertools.utilities.typing import LambdaContext


processor = BatchProcessor(event_type=EventType.SQS)
dynamodb = DynamoDBPersistenceLayer(table_name="workflow-events")
config = IdempotencyConfig(
    event_key_jmespath="messageId",
    use_local_cache=True,
)

logger = Logger(level="INFO",service="workflow_events_track")



@idempotent_function(data_keyword_argument="record", config=config, persistence_store=dynamodb)
def process_workflow_event(record: SQSRecord):
    ... some logic


@logger.inject_lambda_context
@batch_processor(record_handler=process_workflow_event, processor=processor)
def lambda_handler(event: dict, context: LambdaContext):
    config.register_lambda_context(context)  # needed for idempotent functions
    return processor.response()

When the lambda runs though, it raises the following warning....

/var/task/aws_lambda_powertools/utilities/idempotency/persistence/base.py:360: UserWarning: Couldn't determine the remaining time left. Did you call register_lambda_context on IdempotencyConfig?

I'm not sure why that's getting raised if i'm registering the context to the config in the handler. I'm also not sure what the potential impact of this could be. Could records be processed twice?

[heitorlessa]

Hey!

That's our fault - could you create a docs issue so we can address it, please?

Solution

In the meantime, here's how to address it until we correct the docs and think of a better way to handle this more seamlessly within Batch:

• add lambda_context parameter to your record_handler function as described in the Batch docs
• move the config.register.. inside the record handler

Explanation

The decorator @batch_processor will call your record_handler function before it calls the decorated Lambda Handler (final return). Therefore, the Lambda context is only registered in the Idempotency config after all records have been processed.

This doesn't impact your ability to process once, it is working. What it impacts is your ability to recover from half-completed transactions that failed during a timeout incident.

[andrewhharmon]

Thanks! posting the edited code incase anyone else stops by. Yes, I'll open a doc issue.

from typing import Optional
from aws_lambda_powertools.utilities.idempotency import (
    DynamoDBPersistenceLayer, IdempotencyConfig, idempotent_function)
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, batch_processor
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord

from aws_lambda_powertools.utilities.typing import LambdaContext


processor = BatchProcessor(event_type=EventType.SQS)
dynamodb = DynamoDBPersistenceLayer(table_name="workflow-events")
config = IdempotencyConfig(
    event_key_jmespath="messageId",
    use_local_cache=True,
)

logger = Logger(level="INFO",service="workflow_events_track")



@idempotent_function(data_keyword_argument="record", config=config, persistence_store=dynamodb)
def process_workflow_event(record: SQSRecord, lambda_context: Optional[LambdaContext] = None):
    config.register_lambda_context(lambda_context)  # needed for idempotent functions
    ... some logic


@logger.inject_lambda_context
@batch_processor(record_handler=process_workflow_event, processor=processor)
def lambda_handler(event: dict, context: LambdaContext):
    return processor.response()

[heitorlessa]

Thank you so much. I'm flying long-haul for the next 24 hours; I'll send a fix next week.

Thanks again!!

[andrewhharmon]

well, i still seem to be seeing the same error message. hoping i implemented the fix correctly

[heitorlessa]

Allow us to reproduce and we’ll come back with a patch as soon as we can. In-flight Wi-Fi may not be stable so bear with us

…

On Wed, 7 Dec 2022 at 19:29, Andrew Harmon ***@***.***> wrote: well, i still seem to be seeing the same error message. hoping i implemented the fix correctly — Reply to this email directly, view it on GitHub <#1762 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAZPQBE4FTZ3IITQ3VPUN4DWMDJP3ANCNFSM6AAAAAASXAMHQQ> . You are receiving this because you commented.Message ID: <awslabs/aws-lambda-powertools-python/repo-discussions/1762/comments/4336021 @github.com>

[andrewhharmon]

yeah, no hurry. thanks for the quick replies!

[heitorlessa]

Read the source from my phone - it's the Idempotency anti-corruption layer that kicks in before the decorated function gets called hence why it didn't work (sorry).

Using the batch context manager and passing the Lambda context instead of the "batch_processor" decorator works. It's a bit more verbose but we can look into ways to smooth that out later while keeping correctness in place.

...

def lambda_handler(event, context: LambdaContext):
    config.register_lambda_context(context)

    batch = event["Records"]
    with processor(records=batch, handler=record_handler):
        result = processor.process()

    return result

Full sample snippet

from aws_lambda_powertools.utilities.batch import (BatchProcessor, EventType)
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
from aws_lambda_powertools.utilities.idempotency import (
    DynamoDBPersistenceLayer, IdempotencyConfig, idempotent_function)


processor = BatchProcessor(event_type=EventType.SQS)
dynamodb = DynamoDBPersistenceLayer(table_name="idem")
config =  IdempotencyConfig(
    event_key_jmespath="messageId",
    use_local_cache=True,
)


@idempotent_function(data_keyword_argument="record", config=config, persistence_store=dynamodb)
def record_handler(record: SQSRecord):
    # do something with each record 
    return {"message": record["body"]}


def lambda_handler(event, context):
    config.register_lambda_context(context) # guard against sudden Lambda timeouts

    # with Lambda context registered for Idempotency, we can now kick in the Bach processing logic

    batch = event["Records"]
    with processor(records=batch, handler=record_handler):
        # in case you want to access each record processed by your record_handler
        # otherwise ignore the result variable assignment
        result = processor.process()

    return processor.response()

I'm confident it'll work after re-reading the source code; let me know otherwise and I can look into it properly next week when I get my laptop back.

Sent from mobile; apologies for any typos and indentation.

[heitorlessa]

Just merged a docs PR to provide the complete example + better visibility for anyone coming from Batch now. Latest docs will be updated after I complete another unrelated fix.

Thank you again for helping us improve the docs for everyone ;)

[andrewhharmon]

Awesome thanks, I hadn't looked at this since we last chatted but will give it a try this week. BTW, just refactored our SPA to use your Powertools routing for APIGateway. Helped tremendously. Great work over there!