ingest storage: don't return an error when records are successfully i…

…ngested This fixes a flaky TestPartitionReader_Commit because we'd shut down the reader. The bug was introduced in XXX Signed-off-by: Dimitar Dimitrov <[email protected]>
grafana · Oct 2, 2024 · b964ae4 · b964ae4
1 parent 6a998f6
commit b964ae4
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/pkg/storage/ingest/reader.go b/pkg/storage/ingest/reader.go
@@ -503,7 +503,8 @@ func (r *PartitionReader) consumeFetches(ctx context.Context, fetches kgo.Fetche
 		err := consumer.Consume(consumeCtx, records)
 		if err == nil {
 			level.Debug(logger).Log("msg", "closing consumer after successful consumption")
-			break
+			// The context might have been cancelled in the meantime, so we return here instead of breaking the loop and returning the context error
+			return nil
 		}
 		level.Error(logger).Log(
 			"msg", "encountered error while ingesting data from Kafka; should retry",