From 65dd5ac6fbe93d968fe1a4f526e6df577ac915d8 Mon Sep 17 00:00:00 2001 From: chrislu Date: Mon, 20 May 2024 12:27:45 -0700 Subject: [PATCH] fix --- weed/mq/broker/broker_grpc_sub.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/weed/mq/broker/broker_grpc_sub.go b/weed/mq/broker/broker_grpc_sub.go index db2964121..7cbde21f8 100644 --- a/weed/mq/broker/broker_grpc_sub.go +++ b/weed/mq/broker/broker_grpc_sub.go @@ -101,6 +101,7 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.SeaweedMessaging_Subs } imt.AcknowledgeMessage(ack.GetAck().Key, ack.GetAck().Sequence) currentLastOffset := imt.GetOldest() + fmt.Printf("%+v recv (%s,%d), oldest %d\n", partition, string(ack.GetAck().Key), ack.GetAck().Sequence, currentLastOffset) if subscribeFollowMeStream != nil && currentLastOffset > lastOffset { if err := subscribeFollowMeStream.Send(&mq_pb.SubscribeFollowMeRequest{ Message: &mq_pb.SubscribeFollowMeRequest_Ack{ @@ -161,6 +162,8 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.SeaweedMessaging_Subs // reset the sleep interval count sleepIntervalCount = 0 + imt.InflightMessage(logEntry.Key, logEntry.TsNs) + if err := stream.Send(&mq_pb.SubscribeMessageResponse{Message: &mq_pb.SubscribeMessageResponse_Data{ Data: &mq_pb.DataMessage{ Key: logEntry.Key,