一、需求

从指定时间戳（比如 2 小时前）开始消费 Kafka 数据

二、思路

我们知道通过 Kafka 的 API 可以得到指定时间戳对应数据所在的 segment 的起始 offset。那么就可以通过这个功能来粗略的实现需求。

三、实现

我们知道 KafkaUitls.createDirectStream 这个接口可以指定起始点的 offset，那么我们需要做的就变成如下三步：

• 获取 topic 对应的 TopicAndPartitions ，得到当前 topic 有多少 partition
• 从 Kafka 获取每个 partition 指定时间戳所在 segment 的起始 offset
• 将步骤 2 中的 offset 作为参数传入 createDirectStream 即可