Flink·Kafka：为什么我会丢失信息？

Question

我编写了一个非常简单的Flink流媒体作业，它使用Flink Kafka消费者082从Kafka获取数据。

protected DataStream<String> getKafkaStream(StreamExecutionEnvironment env, String topic) {
    Properties result = new Properties();
    result.put("bootstrap.servers", getBrokerUrl());
    result.put("zookeeper.connect", getZookeeperUrl());
    result.put("group.id", getGroup());

        return env.addSource(
                new FlinkKafkaConsumer082<>(
                        topic,
                        new SimpleStringSchema(), result);
}

这工作得很好，每当我在Kafka上将某些内容放入主题时，它都会被我的Flink作业接收并处理。现在我试图看看如果我的Flink作业由于某种原因不在线会发生什么。所以我关闭了flink作业并继续向Kafka发送消息。然后我再次开始我的Flink作业，并期望它会处理同时发送的消息。

然而，我得到了以下信息：

No prior offsets found for some partitions in topic collector.Customer. Fetched the following start offsets [FetchPartition {partition=0, offset=25}]

因此，它基本上忽略了自上次关闭Flink作业以来出现的所有消息，只是在队列末尾开始读取。从我收集的FlinkKafkaConsumer082文档中，它自动负责与Kafka代理同步处理的偏移量。然而，情况似乎并非如此。

我使用单节点Kafka安装（Kafka发行版附带的安装）和单节点Zookeper安装（也是与Kafka发行版捆绑的安装）。

我怀疑这是某种错误配置或类似的东西，但我真的不知道从哪里开始寻找。有人有这个问题，也许解决了？

Answer 1

https://kafka.apache.org/08/configuration.html

设置自动。抵消重置为最小值（默认为最大值）

auto.offset.reset：

当Zookeeper中没有初始偏移或偏移超出范围时，该怎么办：

最小：自动将偏移重置为最小偏移

最大：自动将偏移重置为最大偏移

其他：向消费者抛出异常。

如果将其设置为最大，则当代理上订阅的主题的分区数更改时，使用者可能会丢失一些消息。要防止在添加分区期间丢失数据，请设置auto。抵消重置为最小值

还要确保重启后getGroup（）相同

Answer 2

我找到了原因。您需要在StreamExecutionEnvironment中显式启用检查点，以使Kafka连接器将处理后的偏移量写入Zookeeper。如果不启用它，Kafka连接器将不会写入上次读取的偏移量，因此在重新启动收集作业时，它将无法从那里恢复。所以一定要写：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(); // <-- this is the important part

Anatoly关于更改初始偏移量的建议可能仍然是一个好主意，以防检查点由于某种原因而失败。