Kinesis触发Lambda故障补采记录时EndSequenceNumber匹配问题咨询

认知正确性判断 #

你的原有认知是错误的，拉取到的最后一条记录SequenceNumber无需与告警邮件中的EndSequenceNumber完全匹配，该现象本身不会直接导致记录遗漏，核心原因如下：

• Kinesis Data Stream的序列号是分片内全局递增但不连续的，中间会因为生产者写入失败、分片合并/拆分等操作产生空隙，按条数累计拉取时，第batchSize条记录的序列号自然不一定匹配告警给出的EndSequenceNumber
• Lambda故障告警中给出的EndSequenceNumber是本次故障批次中最后一条有效记录的序列号，而非拉取batchSize条记录后刚好对应的序列号

正确拉取指定范围记录的方案 #

你需要同时基于序列号范围和批次数量设置双终止条件，具体逻辑如下：

1. 使用AT_SEQUENCE_NUMBER类型的分片迭代器，起始位置设为告警给出的StartSequenceNumber
2. 循环调用get_records拉取记录，每次拉取后做两个判断：
   • 累计拉取的记录数是否达到告警给出的batchSize
   • 当前返回的记录中是否存在序列号大于等于告警给出的EndSequenceNumber的记录
3. 任意一个条件满足时立即停止拉取，最后过滤掉所有序列号大于EndSequenceNumber的记录，剩下的就是本次故障批次的完整漏处理记录

参考实现代码 #

import boto3

kinesis_client = boto3.client('kinesis')

def get_failed_batch_records(stream_name, shard_id, start_seq, end_seq, batch_size):
    # 获取起始序列号对应的分片迭代器
    shard_iterator = kinesis_client.get_shard_iterator(
        StreamName=stream_name,
        ShardId=shard_id,
        ShardIteratorType='AT_SEQUENCE_NUMBER',
        StartingSequenceNumber=start_seq
    )['ShardIterator']
    
    collected_records = []
    end_reached = False
    
    while len(collected_records) < batch_size and not end_reached:
        resp = kinesis_client.get_records(
            ShardIterator=shard_iterator, 
            Limit=min(1000, batch_size - len(collected_records))
        )
        # 遍历本次拉取的记录
        for record in resp['Records']:
            if record['SequenceNumber'] > end_seq:
                end_reached = True
                break
            collected_records.append(record)
            if len(collected_records) == batch_size:
                end_reached = True
                break
        # 迭代器失效时终止拉取
        shard_iterator = resp.get('NextShardIterator')
        if not shard_iterator:
            break
    
    return collected_records

额外注意事项 #

• 分片迭代器默认有效期为5分钟，如果批次数据量极大需要长时间拉取，需要重新获取迭代器
• 如果拉取到EndSequenceNumber时累计记录数远小于batch_size，优先检查对应分片是否已经被合并/拆分，确认告警给出的ShardID在拉取时仍然有效
• 处理完记录后可以通过最后一条记录的序列号与EndSequenceNumber比对，只要最后一条序列号小于等于EndSequenceNumber且数量匹配，即为完整的故障批次数据

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