From af19256dbf95e91476bf182203ca5df34fd25ab3 Mon Sep 17 00:00:00 2001 From: chrislu Date: Wed, 29 May 2024 23:34:39 -0700 Subject: [PATCH] skip control messages --- weed/mq/broker/broker_grpc_sub_follow.go | 6 ++++-- weed/mq/client/sub_client/on_each_partition.go | 4 ++++ 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/weed/mq/broker/broker_grpc_sub_follow.go b/weed/mq/broker/broker_grpc_sub_follow.go index eb041dc60..351b904c2 100644 --- a/weed/mq/broker/broker_grpc_sub_follow.go +++ b/weed/mq/broker/broker_grpc_sub_follow.go @@ -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 } diff --git a/weed/mq/client/sub_client/on_each_partition.go b/weed/mq/client/sub_client/on_each_partition.go index 33db81143..14e427713 100644 --- a/weed/mq/client/sub_client/on_each_partition.go +++ b/weed/mq/client/sub_client/on_each_partition.go @@ -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 {