From c9df613b6bfb7e22e4e596151300aefc56ac909f Mon Sep 17 00:00:00 2001 From: chrislu Date: Sun, 31 Mar 2024 00:19:16 -0700 Subject: add publisher name for debugging --- weed/mq/topic/local_partition.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) (limited to 'weed/mq/topic') diff --git a/weed/mq/topic/local_partition.go b/weed/mq/topic/local_partition.go index bff0ec17a..0873d6bd7 100644 --- a/weed/mq/topic/local_partition.go +++ b/weed/mq/topic/local_partition.go @@ -8,6 +8,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb" "github.com/seaweedfs/seaweedfs/weed/util/log_buffer" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "sync" "sync/atomic" "time" @@ -171,7 +173,12 @@ func (p *LocalPartition) MaybeConnectToFollowers(initMessage *mq_pb.PublishMessa for { ack, err := p.followerStream.Recv() if err != nil { - glog.Errorf("Error receiving follower ack: %v", err) + e, _ := status.FromError(err) + if e.Code() == codes.Canceled { + glog.V(0).Infof("local partition %v follower %v stopped", p.Partition, p.follower) + return + } + glog.Errorf("Receiving local partition %v follower %s ack: %v", p.Partition, p.follower, err) return } atomic.StoreInt64(&p.AckTsNs, ack.AckTsNs) -- cgit v1.2.3