This commit is contained in:
Gordon 2022-05-11 19:45:44 +08:00
parent d36f4cf6f8
commit eb7ff77461

View File

@ -141,6 +141,7 @@ func (och *OnlineHistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupS
och.msgHandle[msg.Topic](msg.Value, string(msg.Key))
sess.MarkMessage(msg, "")
if claim.HighWaterMarkOffset()-msg.Offset <= 1 {
log.Debug("", "online msg consume end", claim.HighWaterMarkOffset(), msg.Offset)
och.TriggerCmd(OnlineTopicVacancy)
SetOnlineTopicStatus(OnlineTopicVacancy)
}