From 3148dec1f6314123d63f0b8e8060e13070ae9556 Mon Sep 17 00:00:00 2001 From: chrislu Date: Mon, 20 May 2024 13:29:37 -0700 Subject: [PATCH] avoid concurrent processing for the same key --- weed/mq/broker/broker_grpc_sub.go | 4 ++++ weed/mq/sub_coordinator/inflight_message_tracker.go | 8 ++++++++ 2 files changed, 12 insertions(+) diff --git a/weed/mq/broker/broker_grpc_sub.go b/weed/mq/broker/broker_grpc_sub.go index 7cbde21f8..b4475c525 100644 --- a/weed/mq/broker/broker_grpc_sub.go +++ b/weed/mq/broker/broker_grpc_sub.go @@ -162,6 +162,10 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.SeaweedMessaging_Subs // reset the sleep interval count sleepIntervalCount = 0 + for imt.IsInflight(logEntry.Key) { + time.Sleep(137 * time.Millisecond) + } + imt.InflightMessage(logEntry.Key, logEntry.TsNs) if err := stream.Send(&mq_pb.SubscribeMessageResponse{Message: &mq_pb.SubscribeMessageResponse_Data{ diff --git a/weed/mq/sub_coordinator/inflight_message_tracker.go b/weed/mq/sub_coordinator/inflight_message_tracker.go index f48335435..1b1caca00 100644 --- a/weed/mq/sub_coordinator/inflight_message_tracker.go +++ b/weed/mq/sub_coordinator/inflight_message_tracker.go @@ -67,6 +67,14 @@ func (imt *InflightMessageTracker) GetOldest() int64 { return imt.timestamps.Oldest() } +// IsInflight returns true if the message with the key is inflight. +func (imt *InflightMessageTracker) IsInflight(key []byte) bool { + imt.mu.Lock() + defer imt.mu.Unlock() + _, found := imt.messages[string(key)] + return found +} + // RingBuffer represents a circular buffer to hold timestamps. type RingBuffer struct { buffer []int64