Flink Kafka Producer 中的 Exactly-once 语义

Question

我正在尝试使用 Kafka Source 和 Sink 测试 Flink 的一次性语义：

1. 运行 flink 应用程序，只需将消息从一个主题传输到另一个主题，并行度=1，检查点间隔 20 秒
2. 使用 Python 脚本每 2 秒生成递增整数的消息。
3. 使用 read_commissed 隔离级别的控制台使用者读取输出主题。
4. 手动杀死任务管理器

我希望在输出主题中看到单调递增的整数，无论​​ TaskManager 终止和恢复。

但实际上在控制台消费者输出中看到了一些意想不到的东西：

32
33
34
35
36
37
38
39
40
-- TaskManagerKilled
32
34
35
36
40
41
46
31
33
37
38
39
42
43
44
45

看起来检查点之间的所有消息都在输出主题中重播。这应该是正确的行为还是我做错了什么？

恢复了一张快照： Flink UI

我的弗林克代码：

32
33
34
35
36
37
38
39
40
-- TaskManagerKilled
32
34
35
36
40
41
46
31
33
37
38
39
42
43
44
45

Answer 1

除了将生产者设置为一次性语义之外，您还需要将消费者配置为仅读取来自 kafka 的已提交消息。默认情况下，消费者将读取已提交和未提交的消息。将此设置添加到您的消费者应该会让您更接近您想要的行为。

consumerProperties.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");