AWS Kinesis事件经Lambda送达客户端：判定、移除与重试调度

1. 如何判定事件已成功送达客户端？ #

判定成功的核心是获取客户端的明确业务层确认信号，而非仅依赖网络层发送成功，不同客户端类型的实现方式不同：

• HTTP/REST客户端：Lambda调用客户端接口后，校验返回的2xx状态码，同时要求客户端返回包含事件唯一ID的确认报文，确保客户端已接收并完成业务处理（而非仅收到请求）。
• WebSocket客户端：约定客户端在收到事件后，主动发送带事件ID的ACK消息到Lambda（或后端服务），Lambda仅在收到该ACK后判定送达成功。
• 异步消息客户端（如MQTT、SQS）：依赖客户端的消费确认机制，比如MQTT QoS 1/2级别的PUBACK/PUBCOMP报文，或SQS客户端的消息删除回调通知，以此作为成功送达的依据。

2. 如何手动从Kinesis数据流队列中移除事件？ #

Kinesis通过**检查点（Checkpoint）**标记已消费的事件，手动移除本质是手动提交检查点，步骤如下：

1. 关闭自动检查点：在Lambda的Kinesis事件源配置中，关闭「自动提交检查点」（控制台触发配置或CLI命令aws lambda update-event-source-mapping --uuid <映射ID> --no-enable-auto-checkpoint）。
2. 手动提交检查点：在Lambda代码中，当确认事件成功送达后，调用AWS SDK（v2/v3）的PutCheckpoint接口，传入当前shard ID、成功处理的最后一个事件的序列号，以及消费者ARN（Lambda事件源的消费者ARN）。示例代码（JavaScript v3）：

import { KinesisClient, PutCheckpointCommand } from "@aws-sdk/client-kinesis";

const client = new KinesisClient({});

async function submitCheckpoint(shardId, sequenceNumber, consumerArn) {
  const command = new PutCheckpointCommand({
    ConsumerARN: consumerArn,
    ShardId: shardId,
    SequenceNumber: sequenceNumber
  });
  await client.send(command);
}

注意：Kinesis检查点按shard的序列号位置标记，若批量处理中存在失败事件，仅提交成功事件的最后一个序列号，失败事件会留在流中等待下次消费。

3. 若送达失败，如何调度重试操作？ #

推荐结合SQS延迟队列实现自定义延迟重试，流程清晰且可扩展：

• 步骤1：捕获失败事件：当Lambda推送客户端失败（如超时、非2xx响应、未收到ACK），将事件内容（附带原始事件ID、重试次数、失败原因等元数据）发送到SQS延迟队列，设置DelaySeconds = x*60（x为指定的重试分钟数）。
• 步骤2：移除Kinesis中的事件：提交Kinesis检查点，将该失败事件从Kinesis流中标记为已消费，避免重复处理。
• 步骤3：延迟重试：SQS延迟队列到期后，触发Lambda重新尝试推送客户端；每次重试后更新事件的重试次数，当超过预设的最大重试次数时，将事件转入SQS死信队列，用于后续人工排查。

如果需要更复杂的重试逻辑（如动态调整延迟时间、多阶段重试），可以用AWS Step Functions编排流程，通过Wait状态实现自定义延迟，结合Retry配置控制重试次数和触发条件。

内容的提问来源于stack exchange，提问作者Bear Bile Farming is Torture