问题描述
在这种情况下如何回滚/撤消偏移提交(请参阅 catch 块中的注释)?
public void HandleMessage(ConsumeResult<object,int> messageResult,IConsumer consumer) {
try {
consumer.commit(messageResult);
dbProvider.startTransaction();
//Do db operations
dbProvider.commit();
}
catch (Exception ex) {
dbProvider.rollBack();
//How do I undo kafka commit here?
}
}
基本上,我首先要提交 kafka 偏移量,然后提交到数据库。如果数据库事务失败,我想回滚偏移量。
解决方法
可以向后查找消费者组并提交之前的偏移量,是的。
但是,我认为这与在数据库事务完成后提交没有太大区别