文章目录
- 启用checkpoint
- 禁用checkpoint
- kafka 是否启用enable.auto.commit
- 总结
主要分为三种情况进行讨论,一种是有checkpoint的情况下,一种是没有checkpoint的情况下,还有一种是启用了自动提交的情况下
启用checkpoint
- 这种最简单,offset 保存在checkpoint 中,通过数据一致性的语义不同,offset 的位置也不一样,可能刚好从上一次checkpoint保存的位置开始,也可能位置会再往前一点,会多消费数据。
禁用checkpoint
- 没有了checkpoint ,实际上offset 就取决于enable.auto.commit,如果没有自动提交,那相当于就没有offset 记录,实际的offset 取决于这里的实始化配置,如果有自动提交,就是下一种情况。
可选值
OffsetsInitializer.earliest() - starting from the earliest offsets. This is also the default OffsetsInitializer of the KafkaSource for starting offsets.
OffsetsInitializer.latest() - starting from the latest offsets.
setStartingOffsets()
kafka 是否启用enable.auto.commit
- 如果启用了自动提交,可以通过
OffsetsInitializer.committedOffsets() 设置起始位置,
可选值有public enum OffsetResetStrategy {LATEST, EARLIEST, NONE
}
从已提交的offset的最新,最开始位置提交
setStartingOffsets()
总结
- 注意后面两种情况下,对于开始位置在代码上的设置差异。