bugfix: Stop the current fetcher before replacing it (#9582)

Otherwise, this causes a memory leak as we leave lingering around the resources we used to catch up as quickly as possible as the ingester was starting. part of grafana/mimir-squad#2475 Signed-off-by: gotjosh <[email protected]> (cherry picked from commit 0c60705)
grafana · Oct 10, 2024 · 23603dc · 23603dc
1 parent 245ed65
commit 23603dc
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/pkg/storage/ingest/reader.go b/pkg/storage/ingest/reader.go
@@ -259,6 +259,9 @@ func (r *PartitionReader) switchToOngoingFetcher(ctx context.Context) {
 	}
 
 	if r.kafkaCfg.StartupFetchConcurrency > 0 && r.kafkaCfg.OngoingFetchConcurrency == 0 {
+		// Stop the current fetcher before replacing it.
+		r.fetcher.Stop()
+
 		if r.fetcher == r {
 			// This method has been called before, no need to switch the fetcher.
 			return