主要分为三种情况进行讨论，一种是有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()

总结

• 注意后面两种情况下，对于开始位置在代码上的设置差异。