kafka 重放 重播 从某个时间点或者offset开始消费

2019-02-25  本文已影响4人  把爱放下会走更远
consumer.subscribe(topicA);
consumer.poll(100);//正常订阅topic和poll消息

Set<TopicPartition> assignments = consumer.assignment();//获取consumer所分配的分区信息
Map<TopicPartition, Long> query = new HashMap<>();//构造offsetsForTimes参数,通过时间戳找到offset
for (TopicPartition topicPartition : assignments) {
    System.out.println(topicPartition);
    query.put(topicPartition, 1550804131000L);
}
Map<TopicPartition, OffsetAndTimestamp> result = consumer.offsetsForTimes(query);
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : result.entrySet()) {
    System.out.println(entry);
    consumer.seek(entry.getKey(), entry.getValue().offset());//每个topic的partition都seek到执行的offset
}
                     
上一篇下一篇

猜你喜欢

热点阅读