从 Kafka 消费确切数量的事件答案

【问题标题】：Consuming exact number of events from Kafka从 Kafka 消费确切数量的事件
【发布时间】：2020-11-15 18:26:39
【问题描述】：

当然，流式应用程序是无限制的，但我有一个新的用例，我需要从 Kafka 主题中使用确切数量或更少的消息（可配置，例如 100 条消息）。然后应用程序应该停止。

动机很简单，流量很少使用，不需要实时，所以没有理由拥有一个永久的流媒体应用。

偶尔调用一次应用程序就足够了。

有没有办法使用 FlinkKafkaConsumer 来实现它？

添加一个计数器，当应用程序达到所需的消息数量时将其终止是一种选择，但我更喜欢使用更优雅的东西。

【问题讨论】：

【解决方案1】：

您可以将max.poll.records 设置为100，并将adjust fetch.max.bytes 设置为等于100 条消息大小的数字。例如：1 个混乱 = 10 字节 => fetch.max.bytes = 100*10 字节

【讨论】：

感谢您的回答，几个问题：如果我不知道消息大小是多少，我可以设置 max.poll.records 吗？应用程序会在消耗完作业数后停止吗？如果主题中的消息少于 max.poll.records 值，我该如何自动停止应用程序？
max.poll.records 在没有 fetch.max.bytes 的情况下无法单独工作。如果您无法提供 fetch.max.bytes 的确切数字，则可能会导致无法立即使用确切的消息。对于停止应用程序，我认为检查消息大小不是一个坏主意
所以无论如何我都需要实现计数器，这正是我想要避免的:(
实际上并不是一个每次都在计数的计数器，它只需要在消费者读取消息数量

【解决方案2】：

您可以为 FlinkKafkaConsumer 创建一个包装器 SourceFunction，它委托给它，并在读取目标数量的消息时终止。当一个 Flink 流式作业的所有来源都完成后，工作流将自动停止。

【讨论】：