gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor(scheduler): update log level for kafka message

Browse Source 
@formatter:off

Signed-off-by: JasonHu520 <huzongchao@oppo.com>
This commit is contained in:
JasonHu520 2024-10-16 15:13:32 +08:00 committed by slasher
parent d6c6a775e0
commit d7abb27eac
2 changed files with 2 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
blobstore/scheduler/base/kafka_consumer_group.go
View File

@ -100,7 +100,7 @@ func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai
select {
case message = <-claim.Messages():
if message == nil {
span.Warnf("no message for consume and continue")
span.Debug("no message for consume and continue")
continue
}

2
blobstore/scheduler/blob_deleter.go
View File

@ -515,7 +515,7 @@ func (mgr *BlobDeleteMgr) consume(item *delBlobRet, consumerPause base.ConsumerP
now := time.Now().UTC()
if now.Sub(time.Unix(item.delMsg.Time, 0)) < mgr.safeDelayTime {
sleepDuration := mgr.delayDuration(item.delMsg.Time)
span.Warnf("blob is protected: until[%+v], sleep[%+v]", time.Unix(item.delMsg.Time, 0).Add(mgr.safeDelayTime), sleepDuration)
span.Debugf("blob is protected: until[%+v], sleep[%+v]", time.Unix(item.delMsg.Time, 0).Add(mgr.safeDelayTime), sleepDuration)
ok := sleep(sleepDuration, consumerPause)
if !ok {
item.status = DeleteStatusUndo