Metro 71 fix zombie subscribers to avoid refreshing bindings (#384)

* publish consumer lag as part of ticker * Add GetConsumerLag to IConsumer * Fix lint issues * Add consumer lag metric value * Fix lint issues * Fix lint issues * Fix typo * Added integration test for FetchConsumerLag * Avoid clash with topic separator
razorpay · Jul 5, 2022 · e71f2f5 · e71f2f5
1 parent 58840ee
commit e71f2f5
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/internal/subscriber/subscriber.go b/internal/subscriber/subscriber.go
@@ -142,7 +142,7 @@ func (s *Subscriber) Run(ctx context.Context) {
 		case <-s.healthMonitorTicker.C:
 			lag := s.subscriberImpl.GetConsumerLag()
 			for topic, offset := range lag {
-				tp := strings.Split(topic, "-")
+				tp := strings.Split(topic, "=")
 				if len(tp) == 1 {
 					logger.Ctx(ctx).Errorw("subscriber: failed to parse topic and partition for consumer lag", "topic", topic)
 					continue

diff --git a/pkg/messagebroker/kafka.go b/pkg/messagebroker/kafka.go
@@ -642,7 +642,7 @@ func (k *KafkaBroker) FetchConsumerLag(ctx context.Context) (map[string]uint64,
 	}
 
 	for _, tp := range committed {
-		topicPart := *tp.Topic + "-" + strconv.Itoa(int(tp.Partition))
+		topicPart := *tp.Topic + "=" + strconv.Itoa(int(tp.Partition))
 		low, high, err := k.Consumer.QueryWatermarkOffsets(*tp.Topic, tp.Partition, kafkaMetadataTimeout)
 		if err != nil {
 			logger.Ctx(ctx).Errorw("kafka: failed to fetch watermark offsets",