问题标签 [amazon-kcl]

同步 Kinesis 分片和租约时捕获异常

引起：io.netty.handler.codec.http2.Http2Exception：第一个接收到的帧不是设置。前 5 个字节的十六进制转储：3c21444f43

我想将来自 AWS Kinesis 流的消息处理延迟一小时。我已将 KCL 使用者配置为每四分钟读取一批记录，检查每条记录的时间戳，如果任何记录不到一小时，则停止处理批次，无需检查点。我希望同一个消费者实例每四分钟重新读取一次相同的消息，直到整个批次足够老可以处理，然后对消费者进行检查点。然而，在实践中，消费者只读取一次消息，这意味着它们在准备好处理时被忽略并且不再读取。有没有办法配置消费者每次都重新读取上一个检查点的所有消息？

我们新创建了 kinesis 流，在尝试通过 Java 中的 KCL 连接时会导致以下错误，并且该流可用于动态放置记录。

java.lang.RuntimeException：com.amazonaws.services.kinesis.clientlibrary.exceptions.internal.KinesisClientLibIOException：碎片 [shardId-000000000000] 未关闭。如果我们在重新分片操作正在进行时构建分片列表，就会发生这种情况。

基于文档

它说工作人员将创建 while 并使用描述流来创建租约记录和表。

因此尝试了 aws kinesis describe-stream --stream-name name-of-the-stream 并产生

但是，同一区域下的 dynamo DB 中的租约表缺失。任何建议都会有所帮助。

提前致谢。

我们正在使用kinesis flink connector从 flink 消费和生成记录到 kinesis 中。由于它正在使用KCL，它应该使用dynamoDB它消耗的运动流的偏移量进行条目。我们无法在 中看到任何具有应用程序名称的表dynamoDB。这是预期的行为吗？Flink 连接器版本：1.8 Flink 版本：1.8.0

KCL

ShardConsumerSubscriber:131 - shardId-000000000000: Last request was dispatched at 2020-04-28T12:57:25.166Z, but no response as of 2020-04-28T12:58:00.435Z (PT35.269S). Cancelling subscription, and restarting." But never restarts application and no data is processed after that. Maven dependency used

And the Kinesis configuration

我有一个在 Apache Beam（使用 Spark Runner）中开发的流管道，它从 kinesis 流中读取。

我正在寻找 Apache Beam 中的选项来管理 kinesis 检查点（即定期存储 kinesis 流的当前位置），以便它允许系统从故障中恢复并继续处理流停止的地方。

是否有可供 Apache Beam 支持类似于 Spark Streaming 的 kinesis 检查点的规定（参考链接 - https://spark.apache.org/docs/2.2.0/streaming-kinesis-integration.html）？

我有一个应用程序使用来自 Kinesis 流的记录并进一步处理它们，但性能非常低，所以现在我计划使用 KCL 2.x 迁移到 Kinesis 增强型扇出消费者以提高其性能。由于增强型扇出的 Aws Kinesis 文档非常令人困惑，有人可以帮我举一个例子来说明如何在我的 Java 应用程序中实现此使用者功能吗？

我在 Kinesis 流中的一些特定分片下有一组记录。我正在使用 KCL 2.x 消费者来使用来自 kinesis 的记录，但问题是消费者正在从流中所有可用的分片中获取我的记录。那么有什么方法可以在配置 configBuilder 对象或 KCL 使用者时指定分片或其 ID，以便仅使用来自指定分片的记录。

示例代码：

提前致谢！

我有一个使用 KCL 2.x 来使用来自 Kinesis 的记录的应用程序，存在于不同流分片中的数据具有不同的格式，我想使用具有不同配置的不同 KCL 消费者单独处理它们。例如，如果我在 Kinesis 流中有 3 个分片，我将为不同的分片生成 3 个具有不同配置的不同消费者，即每个分片 1 个消费者。当我创建多个消费者时，我面临的问题是，如果我通过 java 代码配置 3 个不同的消费者，那么它们中的任何一个都占用了所有分片的租约锁定，而其他消费者无法获得该租约。例如，分片总数：3，配置的消费者总数：3，

我如何配置我的 KCL 消费者，以便消费者只租用分配给他的分片。

我正在尝试使用类似于https://github.com/aws-samples/amazon-kinesis-learning的 Kinesis 客户端库来使用 Kinesis 数据流。但在这个例子中，他们安排了这个过程。我想在没有调度程序的情况下使用传入的记录。

我不想使用 DynamoDB、cloudWatch。期望一个简单的消费者在流中消费记录

有没有办法在没有调度程序的情况下使用java处理记录