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

debug

Browse Source
This commit is contained in:
chrislu
2024-05-30 09:15:50 -07:00
parent b485b9695c
commit ff14dfa0ee
2 changed files with 6 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -66,7 +66,8 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.SeaweedMessaging_Subs
} else { } else {
defer func() { defer func() {
println("closing SubscribeFollowMe connection", follower) println("closing SubscribeFollowMe connection", follower)
followerGrpcConnection.Close() subscribeFollowMeStream.CloseSend()
// followerGrpcConnection.Close()
}() }()
followerClient := mq_pb.NewSeaweedMessagingClient(followerGrpcConnection) followerClient := mq_pb.NewSeaweedMessagingClient(followerGrpcConnection)
if subscribeFollowMeStream, err = followerClient.SubscribeFollowMe(ctx); err != nil { if subscribeFollowMeStream, err = followerClient.SubscribeFollowMe(ctx); err != nil {
@@ -94,6 +95,7 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.SeaweedMessaging_Subs
ack, err := stream.Recv() ack, err := stream.Recv()
if err != nil { if err != nil {
if err == io.EOF { if err == io.EOF {
// the client has called CloseSend(). This is to ack the close.
stream.Send(&mq_pb.SubscribeMessageResponse{Message: &mq_pb.SubscribeMessageResponse_Ctrl{ stream.Send(&mq_pb.SubscribeMessageResponse{Message: &mq_pb.SubscribeMessageResponse_Ctrl{
Ctrl: &mq_pb.SubscribeMessageResponse_SubscribeCtrlMessage{ Ctrl: &mq_pb.SubscribeMessageResponse_SubscribeCtrlMessage{
IsEndOfStream: true, IsEndOfStream: true,
@@ -127,6 +129,7 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.SeaweedMessaging_Subs
} }
} }
if lastOffset > 0 { if lastOffset > 0 {
glog.V(0).Infof("saveConsumerGroupOffset %v %v %v %v", t, partition, req.GetInit().ConsumerGroup, lastOffset)
if err := b.saveConsumerGroupOffset(t, partition, req.GetInit().ConsumerGroup, lastOffset); err != nil { if err := b.saveConsumerGroupOffset(t, partition, req.GetInit().ConsumerGroup, lastOffset); err != nil {
glog.Errorf("saveConsumerGroupOffset partition %v lastOffset %d: %v", partition, lastOffset, err) glog.Errorf("saveConsumerGroupOffset partition %v lastOffset %d: %v", partition, lastOffset, err)
} }
@@ -204,6 +207,7 @@ func (b *MessageQueueBroker) getRequestPosition(initMessage *mq_pb.SubscribeMess
return return
} }
if storedOffset, err := b.readConsumerGroupOffset(initMessage); err == nil { if storedOffset, err := b.readConsumerGroupOffset(initMessage); err == nil {
glog.V(0).Infof("resume from saved offset %v %v %v: %v", initMessage.Topic, initMessage.PartitionOffset.Partition, initMessage.ConsumerGroup, storedOffset)
startPosition = log_buffer.NewMessagePosition(storedOffset, -2) startPosition = log_buffer.NewMessagePosition(storedOffset, -2)
return return
} }

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

@@ -42,7 +42,7 @@ func (b *MessageQueueBroker) SubscribeFollowMe(stream mq_pb.SeaweedMessaging_Sub
// Process the received message // Process the received message
if ackMessage := req.GetAck(); ackMessage != nil { if ackMessage := req.GetAck(); ackMessage != nil {
lastOffset = ackMessage.TsNs lastOffset = ackMessage.TsNs
// println("offset", lastOffset) println("sub follower got offset", lastOffset)
} else if closeMessage := req.GetClose(); closeMessage != nil { } else if closeMessage := req.GetClose(); closeMessage != nil {
glog.V(0).Infof("topic %v partition %v subscribe stream closed: %v", initMessage.Topic, initMessage.Partition, closeMessage) glog.V(0).Infof("topic %v partition %v subscribe stream closed: %v", initMessage.Topic, initMessage.Partition, closeMessage)
return nil return nil