基于Lambda与SQS的QLDB Python驱动错误处理方案咨询

我们搭建了一套由SQS触发Lambda函数的QLDB数据导入流程，目标是保障数据管道可靠性，避免因驱动执行失败导致数据无法提交至QLDB而丢失。测试发现Lambda自身故障时，消息会自动重发至队列，但QLDB驱动失败时数据会丢失。已知驱动默认在初始失败后重试4次，现咨询：将qldb_driver.execute_lambda()包裹在try语句中，是否会让驱动正常执行重试逻辑，还是会直接触发except语句进行错误处理？

函数前半部分代码 #

import json
import boto3
import datetime
from pyqldb.driver.qldb_driver import QldbDriver
from utils import upsert, resend_to_sqs, delete_from_sqs

queue_url = 'https://sqs.XXX/'
sqs = boto3.client('sqs', region_name='us-east-1')

ledger = 'XXXXX'
table = 'XXXXX'
qldb_driver = QldbDriver(ledger_name = ledger, region_name='us-east-1')

def lambda_handler(event, context):
    # Simple iterable to identify messages
    i = 0
    
    # Error flag
    error = False
    
    # Empty list to store message send status as well as body or receipt_handle
    batch_messages = []
    
    for record in event['Records']:
        payload = json.loads(record["body"])
        payload['update_ts'] = str(datetime.datetime.now())

        try:
            qldb_driver.execute_lambda(lambda executor: upsert(executor, ledger = ledger, table_name = table, data = payload))

            # If the message sends successfully, give it status 200 and add the recipt_handle to our list 
            # so in case an error occurs later, we can delete this message from the queue.

            message_info = {f'message_{i}': 200, 'receiptHandle': record['receiptHandle']}
            batch_messages.append(message_info)
            
        except Exception as e:
            print(e)

            # Flip error flag to True
            error = True

            # If the commit fails, set status 400 and add the message's body to our list.
            # This will allow us to send the message back to the queue during error handling.

            message_info = {f'message_{i}': 400, 'body': record['body']}
            batch_messages.append(message_info)
        
        i += 1

后续错误处理流程代码 #

# Begin error handling
    if error:
        count = 0
        for j in range(len(batch_messages)):
            # If a message was sent successfully delete it from the queue
            if batch_messages[j][f'message_{j}'] == 200:
                receipt_handle = batch_messages[j]['receiptHandle']
                delete_from_sqs(sqs, queue_url, receipt_handle)
            
            # If the message failed to commit to QLDB, send it back to the queue
            else:
                body = batch_messages[j]['body']
                resend_to_sqs(sqs, queue_url, body)
                count += 1
                
        print(f"ERROR(S) DETECTED - {count} MESSAGES RETURNED TO QUEUE")
                
    else:
        print("BATCH PROCESSING SUCCESSFUL")

解答 #

将qldb_driver.execute_lambda()放在try/except块中不会打断驱动的默认重试逻辑。

PyQLDB驱动的execute_lambda()方法内部已封装默认4次的重试机制，只有当所有重试尝试均失败后，才会抛出异常触发except代码块：

• 若首次执行失败但后续重试成功，execute_lambda()会正常返回，不会进入except分支，你可以正常标记消息为成功并后续删除。
• 仅当4次重试全部失败时，才会抛出异常进入except分支，此时你标记消息失败并重发回SQS的逻辑才会触发。

额外优化建议 #

1. 统一时间戳格式：当前用str(datetime.datetime.now())生成的时间戳格式不统一且未指定时区，建议改用UTC时区的ISO标准格式：

   payload['update_ts'] = datetime.datetime.now(datetime.timezone.utc).isoformat()
   

2. 简化消息状态存储结构：当前用message_{i}作为状态键的方式易因索引不一致出错，建议改用固定键名：

   # 成功时
   message_info = {'status': 200, 'receiptHandle': record['receiptHandle']}
   # 失败时
   message_info = {'status': 400, 'body': record['body']}
   

   后续遍历判断时直接使用batch_messages[j]['status'] == 200即可，逻辑更清晰且不易出错。

内容的提问来源于stack exchange，提问作者acswan9690