gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

skip control messages

Browse Source
This commit is contained in:
chrislu
2024-05-29 23:34:39 -07:00
parent ad2b878130
commit af19256dbf
2 changed files with 8 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
weed/mq/broker/broker_grpc_sub_follow.go
View File

@@ -53,9 +53,11 @@ func (b *MessageQueueBroker) SubscribeFollowMe(stream mq_pb.SeaweedMessaging_Sub
t, p := topic.FromPbTopic(initMessage.Topic), topic.FromPbPartition(initMessage.Partition)
err = b.saveConsumerGroupOffset(t, p, initMessage.ConsumerGroup, lastOffset)
if lastOffset > 0 {
err = b.saveConsumerGroupOffset(t, p, initMessage.ConsumerGroup, lastOffset)
}
glog.V(0).Infof("shut down follower for %v", initMessage)
glog.V(0).Infof("shut down follower for %v offset %d", initMessage, lastOffset)
return err
}

4
weed/mq/client/sub_client/on_each_partition.go
View File

@@ -93,6 +93,10 @@ func (sub *TopicSubscriber) onEachPartition(assigned *mq_pb.BrokerPartitionAssig
}
switch m := resp.Message.(type) {
case *mq_pb.SubscribeMessageResponse_Data:
if m.Data.Ctrl != nil {
glog.V(2).Infof("subscriber %s received control from producer:%s isClose:%v", sub.SubscriberConfig.ConsumerGroup, m.Data.Ctrl.IsClose)
continue
}
executors.Execute(func() {
processErr := sub.OnEachMessageFunc(m.Data.Key, m.Data.Value)
if processErr == nil {