为什么关闭分片时 Amazon Kinesis 流上需要检查点？答案

【问题标题】：Why checkpoint is needed on an Amazon Kinesis stream when shutting down a shard?为什么关闭分片时 Amazon Kinesis 流上需要检查点？
【发布时间】：2015-03-09 18:22:41
【问题描述】：

将一个分片分成 2 个子分片时，父分片将关闭。预计记录处理器（此处使用 KCL）会在发生这种情况时检查点，如以下 KCL 源代码所示：

try {
                recordProcessor.shutdown(recordProcessorCheckpointer, reason);
                String lastCheckpointValue = recordProcessorCheckpointer.getLastCheckpointValue();
                if (reason == ShutdownReason.TERMINATE) {
                    if ((lastCheckpointValue == null)
                            || (!lastCheckpointValue.equals(SentinelCheckpoint.SHARD_END.toString()))) {
                        throw new IllegalArgumentException("Application didn't checkpoint at end of shard "
                                + shardInfo.getShardId());
                    }
                }

问题是：

这个检查点必不可少吗？
如果记录处理器没有检查点并吸收异常会发生什么？

我问的原因是因为在我的用例中，我想确保流中的每条记录都已处理到 s3，现在如果分片关闭，可能有尚未刷新的项目，因此我想确保他们会对子碎片的新消费者/工人感到不满？

如果我检查站，他们不会反感。

有什么想法吗？

提前致谢。

【问题讨论】：

标签： amazon-kinesis

【解决方案1】：

物品不会在碎片之间移动。重新分片后，新记录被放入新分片，但旧记录永远不会从父分片转移，并且不再向（现已关闭的）父分片添加新记录。即使在父分片关闭后，数据也会在其正常的 24 小时生命周期内保留在父分片中。您的记录处理器只有在到达父分片的数据末尾后才会关闭。

http://docs.aws.amazon.com/kinesis/latest/dev/kinesis-using-sdk-java-after-resharding.html

顺便说一句，您可能知道 SDK API 很难，而且客户端库也好不到哪里去。试试连接器库，它是一个更好的 API，包含一个 S3 归档应用程序示例。

https://github.com/awslabs/amazon-kinesis-connectors

【讨论】：

谢谢。这很有用。根据您所说的一个问题，如果我将 1000 条记录放在一个分片流中。然后在消耗了 500 个之后，我将流分成 2 个。假设我使用 KCL（不使用本机 API 读取），KCL 的行为将是什么。它会确保我将有 3 个 RecordProcessor 一段时间，直到剩余的 500 个被消耗完，还是只有 2 个用于子分片的记录处理器，基本上剩下的 500 个都丢失了？提前谢谢。
您的消费对记录的处理没有任何影响。在输入时选择分片。如果将 1000 条记录放在一个分片中，然后拆分，则原始分片中将有 1000 条记录，而子分片中将没有记录。什么时候消费都没关系。
澄清一下，库将为子分片生成另外两个处理器，但在添加更多记录之前它们不会执行任何操作。所有 1000 条记录都将由原始父分片处理器处理，因为那是这些记录所在的位置。
对不起，我没说清楚，真实消费没有影响，这只是潜在数据丢失的一个例子。我会澄清更多，当一个分片被分割时，我清楚地看到我的记录处理器上调用了关闭方法。我将此关闭解释为：“不再向您发送记录”。问题是：是否保证 KCL 在将所有未决记录发送到我的处理器后会调用关闭方法，还是会在拆分时立即调用关闭，在这种情况下，我基本上丢失了自处理器正在关闭？
当到达关闭的分片数据结束时（当分片迭代器返回 null 时）你被关闭，当分片关闭时不是。您不会丢失记录。