[POP]PopReviveService在使用流行模式消费时会重复消费reviveTopic。

重复打印以下日志，这意味着重复使用reviveTopic的消息。

2021-10-29 15:51:12 INFO PopReviveService_0 - reviveQueueId=0,find ack, offset:142765, raw : {"ackOffset":283798,"consumerGroup":"pop-consumer-1","popTime":1635476075611,"queueId":0,"startOffset":283791,"topic":"%RETRY%pop-consumer-1_-topic"}

1、 当popReviveService消耗reviveTopic时，当消息列表没有ck消息时，它无法提交消耗偏移。相关代码是PopReviveService.java中的第365-371行

2、 此外，为什么newOffset是由ck消息决定的IMO，这应该是popReviveService所消耗的真正抵消。相关代码如下。

long newOffset = consumeReviveObj.oldOffset; for (PopCheckPoint popCheckPoint : sortList) { if (!checkAndSetMaster()) { POP_LOGGER.info("slave skip ck process , revive topic={}, reviveQueueId={}", reviveTopic, queueId); break; } if (consumeReviveObj.endTime - popCheckPoint.getReviveTime() <= (PopAckConstants.ackTimeInterval + PopAckConstants.SECOND)) { break; }

// check normal topic, skip ck , if normal topic is not exist
String normalTopic = KeyBuilder.parseNormalTopic(popCheckPoint.getTopic(), popCheckPoint.getCId());
if (brokerController.getTopicConfigManager().selectTopicConfig(normalTopic) == null) {
    POP_LOGGER.warn("reviveQueueId={},can not get normal topic {} , then continue ", queueId, popCheckPoint.getTopic());
    newOffset = popCheckPoint.getReviveOffset();   //Is this right? 
    continue;
}
if (null == brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(popCheckPoint.getCId())) {
    POP_LOGGER.warn("reviveQueueId={},can not get cid {} , then continue ", queueId, popCheckPoint.getCId());
    newOffset = popCheckPoint.getReviveOffset();   
    continue;
}

reviveMsgFromCk(popCheckPoint);

newOffset = popCheckPoint.getReviveOffset();  //and this line

}

原提问者GitHub用户cserwen