配置 Kafka Consumer 开始消费的位置

Flink Kafka Consumer 允许通过配置来确定 Kafka 分区的起始位置。

Flink Kafka Consumer 的所有版本都具有上述明确的起始位置配置方法。

• setStartFromGroupOffsets（默认方法）：从 Kafka brokers 中的 consumer 组（consumer 属性中的 group.id 设置）提交的偏移量中开始读取分区。如果找不到分区的偏移量，那么将会使用配置中的 auto.offset.reset 设置。

• setStartFromEarliest() 或者 setStartFromLatest()：从最早或者最新的记录开始消费，在这些模式下，Kafka 中的 committed offset 将被忽略，不会用作起始位置。

• setStartFromTimestamp(long)：从指定的时间戳开始。对于每个分区，其时间戳大于或等于指定时间戳的记录将用作起始位置。如果一个分区的最新记录早于指定的时间戳，则只从最新记录读取该分区数据。在这种模式下，Kafka 中的已提交 offset 将被忽略，不会用作起始位置。

• setStartFromSpecificOffsets：从指定的分区的offset位置开始读取，如指定的offsets中不存在某个分区，该分区从group offset位置开始读取

例子如下：为每个分区指定 consumer 应该开始消费的具体 offset：

上面的例子中使用的配置是指定从 myTopic 主题的 0 、1 和 2 分区的指定偏移量开始消费。offset 值是 consumer 应该为每个分区读取的下一条消息。

请注意：如果 consumer 需要读取在提供的 offset 映射中没有指定 offset 的分区，那么它将回退到该特定分区的默认组偏移行为（即 setStartFromGroupOffsets()）。

请注意：当 Job 从故障中自动恢复或使用 savepoint 手动恢复时，这些起始位置配置方法不会影响消费的起始位置。在恢复时，每个 Kafka 分区的起始位置由存储在 savepoint 或 checkpoint 中的 offset 确定（有关 checkpointing 的信息，请参阅下一节，以便为 consumer 启用容错功能）。

Kafka Consumer 和容错

伴随着启用 Flink 的 checkpointing 后，Flink Kafka Consumer 将使用 topic 中的记录，并以一致的方式定期检查其所有 Kafka offset 和其他算子的状态。如果 Job 失败，Flink 会将流式程序恢复到最新 checkpoint 的状态，并从存储在 checkpoint 中的 offset 开始重新消费 Kafka 中的消息。

因此，设置 checkpoint 的间隔定义了程序在发生故障时最多需要返回多少。

为了使 Kafka Consumer 支持容错，需要在 执行环境 中启用拓扑的 checkpointing。

如果未启用 checkpoint，那么 Kafka consumer 将定期向 Zookeeper 提交 offset。