From 835451e331ed37ad2089f94dcb4bd124a66d9ac6 Mon Sep 17 00:00:00 2001 From: yukun Date: Mon, 22 Nov 2021 18:51:14 +0800 Subject: [PATCH] Update rocksmq consume slowly log (#12179) Signed-off-by: fishpenguin --- internal/util/rocksmq/server/rocksmq/rocksmq_impl.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/internal/util/rocksmq/server/rocksmq/rocksmq_impl.go b/internal/util/rocksmq/server/rocksmq/rocksmq_impl.go index 9edf15af70..962e3be8ef 100644 --- a/internal/util/rocksmq/server/rocksmq/rocksmq_impl.go +++ b/internal/util/rocksmq/server/rocksmq/rocksmq_impl.go @@ -589,6 +589,7 @@ func (rmq *rocksmq) Consume(topicName string, groupName string, n int) ([]Consum } lock.Lock() defer lock.Unlock() + getLockTime := time.Since(start).Milliseconds() metaKey := constructCurrentID(topicName, groupName) currentID, err := rmq.kv.Load(metaKey) @@ -669,9 +670,11 @@ func (rmq *rocksmq) Consume(topicName string, groupName string, n int) ([]Consum } go rmq.updateAckedInfo(topicName, groupName, consumedIDs) - log.Debug("Rocksmq produce successfully ", zap.String("topic", topicName), - zap.String("groupName", groupName), - zap.Int64("elapsed", time.Since(start).Milliseconds())) + getConsumeTime := time.Since(start).Milliseconds() + if getLockTime > 200 || getConsumeTime > 200 { + log.Warn("rocksmq consume too slowly", zap.String("topic", topicName), + zap.Int64("get lock elapse", getLockTime), zap.Int64("consume elapse", getConsumeTime)) + } return consumerMessage, nil }