fix(dataobj): Fix shutdown race in dataobj consumer (#16157)

grafana · Feb 10, 2025 · 656b1fa · 656b1fa
1 parent 82cfaea
commit 656b1fa
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/pkg/dataobj/consumer/service.go b/pkg/dataobj/consumer/service.go
@@ -109,6 +109,10 @@ func (s *Service) handlePartitionsAssigned(ctx context.Context, client *kgo.Clie
 
 func (s *Service) handlePartitionsRevoked(partitions map[string][]int32) {
 	level.Info(s.logger).Log("msg", "partitions revoked", "partitions", formatPartitionsMap(partitions))
+	if s.State() == services.Stopping {
+		// On shutdown, franz-go will send one more partitionRevoked event which we need to ignore to shutdown gracefully.
+		return
+	}
 	s.partitionMtx.Lock()
 	defer s.partitionMtx.Unlock()