带有MultiLangDaemon的AWS KCL：检查指向“最新”？

Question

我有一个Kinesis消费者，他的工作是跟踪系统中的“当前活跃用户”。用户每分钟都会向一个Kinesis流发送一个心跳信号，这个系统只会保存一个它所看到的所有唯一用户GUID的列表，以及他们上次从该GUID接收心跳信号的时间。如果2分钟内没有检测到心跳信号，我们假设该用户不再处于活动状态，并将其从“当前活动用户”列表中逐出。非常直截了当。

因为这个系统只关注当前活跃的用户，所以我们不需要回处理旧消息。如果我们将此消费者关闭2小时，然后再将其打开，我们希望从“最新”消息开始处理，而不是从我们停止的地方开始。

最后，根据Amazon Kinesis Client NodeJS示例，这已实现为NodeJS应用程序，使用MultiLangDaemon与Kinesis客户端库进行html" target="_blank">通信。

在正常使用下，我发现始终从“LATEST”恢复的最佳方法是永远不要使用KCL的检查点功能。例如，在我的进程记录方法的底部，我有以下内容：

    // We don't checkpoint with kinrta, because if we crash for some reason we
    // want to immediately catch back up to live instead of wasting time
    // processing expired heartbeats
    // processRecordsInput.checkpointer.checkpoint(sequenceNumber,
      // function(err, checkpointedSequenceNumber) {

        completeCallback();

      // }
    // );

这样，每当我杀死消费者并重新启动它时，它都会查看<代码>*。属性文件，并看到“initialPositionInStream”是“最新的”，然后从那里开始处理。

然而

当我重新分片流（分割分片或合并分片）时，我遇到了一个问题。当我重新分片时，新分片上的检查点没有设置为“最新”，而是设置为“TRIM_HORIZON”。因为我从来没有重新检查点，这意味着如果我的消费者关闭并重新启动，我最终不得不处理24小时的数据。

我可以通过编辑KCL用于管理检查点的Dynamo表来手动修复此问题，但这显然不是一个可扩展的解决方案。我尝试使用检查指针并传递字符串“LATEST”而不是序列号，但这会引发序列号无效的错误。

我如何告诉KCL，当我重新切分时，我想在新切分上将检查点设置为“最新”？

作为一个简单的解决方案，我考虑过只使用DynamoDBSDK并修复初始化方法中的检查点。它很难看，但我认为它会起作用（假设Amazon不改变他们管理KCL表的方式）

根据所描述的“hack-y解决方案”，我编写了以下小助手方法：

/**
 * Assumes the current shardId (available in the initialize method's
 * `initializeInput.shardId`) is stored in the global "state" object,
 * accessible via the "state" import
 */

import { Kinesis, DynamoDB } from "aws-sdk";
import state from "../state";
import logger from "./logger";
 
const kinesis = new Kinesis();
const ddb = new DynamoDB.DocumentClient();

const log = logger().getLogger("recordProcessor");
const appName = process.env.APP_NAME;

export default async function (startingCheckpoint: string) { 
    // We can't update any Dynamo tables if we don't know which table to update
    if (!appName) return;

    // Compute the name of the shard JUST BEFORE ours
    // Because Kinesis uses an "exclusive" start ID...
    const shardIdNum = parseInt(state.shardId.split("-")[1]) - 1;
    const startShardId = "shardId-" + ("000000000000" + shardIdNum).substr(-12);

    // Pull data about our current shard
    const kinesisResp = await kinesis.listShards({
        StreamName: process.env.KINESIS_STREAM_NAME,
        MaxResults: 1,
        ExclusiveStartShardId: startShardId
    }).promise();
    const oldestSeqNumber = kinesisResp.Shards[0].SequenceNumberRange.StartingSequenceNumber;

    // Pull data about our current checkpoint
    const dynamoResp = await ddb.get({
        TableName: appName,
        Key: {
            leaseKey: state.shardId
        }
    }).promise();
    const prevCheckpoint = dynamoResp.Item.checkpoint;

    log.debug(`Oldest sequence number in Kinesis shard: ${oldestSeqNumber} vs checkpoint: ${prevCheckpoint}`);

    // Determine if we need to "fix" anything
    if (startingCheckpoint === "TRIM_HORIZON") {

        // If our checkpoint is before the oldest sequence number, reset it to
        // "TRIM_HORIZON" so we pull the oldest sequence number
        if (prevCheckpoint < oldestSeqNumber) {
            log.info("Updating checkpoint to TRIM_HORIZON");

            await ddb.update({
                TableName: appName,
                Key: {
                    leaseKey: state.shardId
                },
                UpdateExpression: "SET #checkpoint = :value",
                ExpressionAttributeNames: {
                    "#checkpoint": "checkpoint"
                },
                ExpressionAttributeValues: {
                    ":value": "TRIM_HORIZON"
                }
            }).promise();
        }

    } else if (startingCheckpoint === "LATEST") {

        if (prevCheckpoint !== "LATEST") {
            log.info("Updating checkpoint to LATEST");

            await ddb.update({
                TableName: appName,
                Key: {
                    leaseKey: state.shardId
                },
                UpdateExpression: "SET #checkpoint = :value",
                ExpressionAttributeNames: {
                    "#checkpoint": "checkpoint"
                },
                ExpressionAttributeValues: {
                    ":value": "LATEST"
                }
            }).promise();
        }

    } else {
        log.warn("We can't 'fix' checkpoints that aren't TRIM_HORIZON or LATEST");
    }
}

我进行了测试，这正确有效地更新了DynamoDB表，但它不会立即开始从新位置提取记录。看起来KCL在调用initialize方法之前读取了一次检查点，并且从不重新读取。

此时，我正在寻找一种方法来告诉KCL“开始使用新的检查点”，或者一种方法来优雅地重新启动消费者，以便它重新初始化所有内容。这两样我都没有，但我会继续研究的。也许我可以在MultiLangDaemon文档中找到一些我可以写入STDOUT的东西。。。

Answer 1

经过大量研究，我得出结论，亚马逊没有提供任何方法来要求优雅的关闭。您只需使消费者崩溃（进程.退出（）），然后等待Docker重新启动它。

然而，在我的hack-y“checkpoint fixer”脚本（我在initialize（）回调中运行）和这个hack-y“crash to restart”方法之间，我现在有了一个适当更新我的检查点的解决方案，所以现在动觉对我来说运行更加平稳。