如何解决如何在Kafka侦听器方法中寻找特定的偏移量?
我正在尝试使用我的kafka侦听器方法从SQL数据库中寻求偏移量。 我在代码中使用了registerSeekCallback方法,但是在运行使用者(或启动容器)时会调用此方法。假设我的使用者正在运行,并且在MySql数据库中最后提交的偏移量为20。我手动将Mysql数据库中最后提交的偏移量更改为11,但是除非重新启动我的使用者(容器已重新启动),否则我的使用者将继续从21读取数据。我正在寻找任何选项,如果我可以在侦听器方法本身中覆盖或查找偏移量。任何帮助将不胜感激。
public class Listen implements ConsumerSeekAware
{
@Override
public void registerSeekCallback(ConsumerSeekCallback callback)
{
// fetching offset from a database
Integer offset = offsetService.getOffset();
callback.seek("topic-name",offset);
}
@KafkaListener(topics = "topic-name",groupId = "group")
public void listen(ConsumerRecord record Acknowledgment acknowledgment) throws Exception
{
// processing the record
acknowledgment.acknowledge(); //manually commiting the record
// committing the offset to MySQL database
}
}
使用新的侦听器方法进行编辑:-
@KafkaListener(topics = "topic-name",groupId = "group")
public void listen(ConsumerRecord record Acknowledgment acknowledgment,@Header(KafkaHeaders.CONSUMER) Consumer<?,?> consumer)) throws Exception {
// seeking old offset stored in database (which is 11 )
consumer.seek(partition,offsetService.getOffset());
log.info("record offset is {} and value is {}",record.offset(),record.value() );
acknowledgment.acknowledge();
}
在数据库中,我最后一次提交的偏移量是11,kafka结束时的最后提交的偏移量是21。当我在kafka主题中写了一条新记录(即在offset 22上)时,我的使用者首先触发并处理22 offset,然后返回查找偏移量11并从那里开始处理。 为什么我在寻求偏移量11时为什么先消耗偏移量22?
使用上面的代码,每次我向kafka顶部写入新消息时,它首先处理记录,然后查找数据库中存在的偏移量。我有什么办法可以避免这种情况?
解决方法
Locator Strategies中有几种技术。
请记住,在下一次民意调查之前,对消费者执行搜索将不会生效(在上次民意调查中获取的所有记录都将首先发送给消费者)。
编辑
这是一个例子:
@SpringBootApplication
public class So63429201Application {
public static void main(String[] args) {
SpringApplication.run(So63429201Application.class,args).close();
}
@Bean
public ApplicationRunner runner(KafkaTemplate<String,String> template,Listener listener) {
return args -> {
IntStream.range(0,10).forEach(i -> template.send("so63429201",i % 3,null,"foo" + i));
Thread.sleep(8000);
listener.seekToTime(System.currentTimeMillis() - 11000);
Thread.sleep(8000);
listener.seekToOffset(new TopicPartition("so63429201",0),11);
Thread.sleep(8000);
};
}
@Bean
public NewTopic topic() {
return TopicBuilder.name("so63429201").partitions(3).replicas(1).build();
}
}
@Component
class Listener extends AbstractConsumerSeekAware {
@KafkaListener(id = "so63429201",topics = "so63429201",concurrency = "2")
public void listen(String in) {
System.out.println(in);
}
@Override
public void onPartitionsAssigned(Map<TopicPartition,Long> assignments,ConsumerSeekCallback callback) {
System.out.println(assignments);
super.onPartitionsAssigned(assignments,callback);
callback.seekToBeginning(assignments.keySet());
}
public void seekToTime(long time) {
getSeekCallbacks().forEach((tp,callback) -> callback.seekToTimestamp(tp.topic(),tp.partition(),time));
}
public void seekToOffset(TopicPartition tp,long offset) {
getSeekCallbackFor(tp).seek(tp.topic(),offset);
}
}
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。