Kinesis Stream数据丢失排查：多消费者未接收全生产数据

你的Kinesis消费者出现数据丢失的问题，核心是代码逻辑对Kinesis的迭代器机制理解有误，咱们一步步拆解原因和修复方案：

核心原因1：ShardIteratorType: 'LATEST'的使用逻辑错误 #

LATEST迭代器的特性是仅读取调用getShardIterator之后新写入Kinesis的记录，在此之前的所有历史记录都会被直接跳过。

看你的消费者代码，getRecord函数每隔1秒就会重新执行全流程：重新查询shard列表、为每个shard创建新的LATEST迭代器、单次调用getRecords。这就导致：

• 生产者发送的a、b、c，都是在消费者创建某次迭代器之前写入的，自然会被LATEST迭代器忽略；
• 只有d刚好是在某次迭代器创建之后发送的，才被读取到。

核心原因2：没有复用NextShardIterator延续读取进度 #

Kinesis的getRecords返回结果里会包含NextShardIterator，这是用来读取下一批记录的关键凭证。但你的代码每次都重新生成全新的迭代器，完全没有延续之前的读取进度，相当于每次都从“最新位置”重新开始读，中间的记录直接被跳过。

核心原因3：独立消费者的重复读取逻辑（非直接丢数据，但浪费资源） #

你启动了3个独立消费者，每个都会遍历所有shard尝试读取。如果你的测试stream只有1个shard（从结果看大概率是这样），三个消费者其实都在读取同一个shard，但因为都是用LATEST迭代器，它们只会拿到各自创建迭代器之后的新记录，无法实现协同消费。

修复方案 #

1. 复用NextShardIterator，持续读取shard #

修改消费者代码，为每个shard保存对应的迭代器，而非每次重新创建，示例思路如下：

// 保存每个shard的迭代器，维护读取进度
const shardIterators = {};

// 初始化所有shard的迭代器
function initShardIterators() {
  kinesis.describeStream({ StreamName: 'test-stream1' }, function(err, streamData) {
    if (err) {
      console.log(err, err.stack);
      return;
    }
    streamData.StreamDescription.Shards.forEach(shard => {
      kinesis.getShardIterator({
        ShardId: shard.ShardId,
        ShardIteratorType: 'LATEST', // 若需读取历史记录，改为TRIM_HORIZON
        StreamName: 'test-stream1'
      }, function(err, shardIteratordata) {
        if (!err) {
          shardIterators[shard.ShardId] = shardIteratordata.ShardIterator;
          // 初始化完成后开始持续读取该shard
          readShard(shard.ShardId);
        }
      });
    });
  });
}

// 持续读取单个shard的记录
function readShard(shardId) {
  const iterator = shardIterators[shardId];
  if (!iterator) return;

  kinesis.getRecords({ ShardIterator: iterator }, function(err, recordsData) {
    if (err) {
      console.log(err, err.stack);
      // 出错后重新初始化迭代器
      setTimeout(() => initShardIterators(), 1000);
      return;
    }

    // 处理收到的记录
    recordsData.Records.forEach(record => {
      console.log(record.Data.toString(), shardId);
    });

    // 更新迭代器，为下一次读取做准备
    shardIterators[shardId] = recordsData.NextShardIterator;
    // 继续读取下一批记录
    setTimeout(() => readShard(shardId), 1000);
  });
}

// 初始化一次后，持续运行读取逻辑
initShardIterators();

2. 选择合适的迭代器类型 #

• 若需要读取所有历史记录（包括已丢失的a、b、c），将ShardIteratorType改为TRIM_HORIZON，它会从shard的起始位置开始读取所有记录；
• 若仅需读取迭代器创建后的新记录，LATEST是合适的，但必须保证迭代器是持续复用的，而非每次重新创建。

3. 使用Kinesis Client Library（KCL）简化消费逻辑 #

手动管理shard迭代器和读取进度极易出错，AWS官方的KCL会自动处理：

• Shard的分配与负载均衡（多个消费者自动分配不同shard）
• 迭代器的保存与续读
• Checkpointing（记录读取进度，重启后从上次位置继续）
  这能大幅降低手动代码的出错概率。

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