ingest storage: add cleanup to TestPartitionReader_ConsumeAtStartup (#9579)

* ingest storage: add cleanup to TestPartitionReader_ConsumeAtStartup

related to XXX; this PR fixes the race on `testing.T.log()` mentioned in that issue. We don't wait for the fetchers to exit and as a result they use the logger after the parent test has exited.

Signed-off-by: Dimitar Dimitrov <dimitar.dimitrov@grafana.com>

* Expect errors when shutting down PartitionReadr

Signed-off-by: Dimitar Dimitrov <dimitar.dimitrov@grafana.com>

---------

Signed-off-by: Dimitar Dimitrov <dimitar.dimitrov@grafana.com>

Author: dimitarvdimitrov

pkg/storage/ingest/reader_test.go

@@ -1305,6 +1305,11 @@ func TestPartitionReader_ConsumeAtStartup(t *testing.T) {
 
 		readerCtx, cancelReaderCtx := context.WithCancel(ctx)
 		require.NoError(t, reader.StartAsync(readerCtx))
+		t.Cleanup(func() {
+			// Interrupting startup should fail the service.
+			// A context cancellation error shouldn't be swallowed and interpreted as "startup went ok"
+			assert.ErrorIs(t, services.StopAndAwaitTerminated(ctx, reader), context.Canceled)
+		})
 
 		// Wait until the Kafka cluster received at least 1 ListOffsets request.
 		test.Poll(t, 5*time.Second, true, func() interface{} {
@@ -1349,6 +1354,11 @@ func TestPartitionReader_ConsumeAtStartup(t *testing.T) {
 
 		readerCtx, cancelReaderCtx := context.WithCancel(ctx)
 		require.NoError(t, reader.StartAsync(readerCtx))
+		t.Cleanup(func() {
+			// Interrupting startup should fail the service.
+			// A context cancellation error shouldn't be swallowed and interpreted as "startup went ok"
+			assert.ErrorIs(t, services.StopAndAwaitTerminated(ctx, reader), context.Canceled)
+		})
 
 		// Wait until the Kafka cluster received at least 1 Fetch request.
 		test.Poll(t, 5*time.Second, true, func() interface{} {