[chore][pkg/stanza] Fixed goleak errors on recombine benchmarks (open-telemetry#43117)

gnak-yar · tommyers-elastic · commit ab70032d79fa · 2025-10-10T11:33:48.000+01:00
#### Description Fixed one of the errors reported in open-telemetry#43044 Two goleak errors I faced from the `recombine` package. One of the errors indicates that `flushLoop` is still running on a goroutine after test finished. This is due to the benchmark tests never call operator's `Stop()`(More precisely, `Transformer.chClose` is never closed), so `flushLoop()` indefinitely runs. Calling `op.Stop()` makes it silent. ``` goleak: Errors on successful test run: found unexpected goroutines: [Goroutine 61 in state select, with github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine.(*Transformer).flushLoop on top of the stack: github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine.(*Transformer).flushLoop(0x14000188a80) /Users/ray.kang/github.com/gnak-yar/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine/transformer.go:62 +0x64 created by github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine.(*Transformer).Start in goroutine 104 /Users/ray.kang/github.com/gnak-yar/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine/transformer.go:56 +0x5c ``` The other error occurs because there's an infinite loop receiving from a channel of `FakeOutput`. Closing the channel resolves the error. ``` Goroutine 103 in state chan receive, with github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine.BenchmarkRecombine.func1 on top of the stack: github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine.BenchmarkRecombine.func1() /Users/ray.kang/github.com/gnak-yar/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine/transformer_test.go:795 +0x2c created by github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine.BenchmarkRecombine in goroutine 21 /Users/ray.kang/github.com/gnak-yar/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/recombine/transformer_test.go:793 +0x498 ```  #### Link to tracking issue Fixes open-telemetry#43044  #### Testing Unit tests
diff --git a/pkg/stanza/operator/transformer/recombine/transformer_test.go b/pkg/stanza/operator/transformer/recombine/transformer_test.go
@@ -791,8 +791,8 @@ func BenchmarkRecombine(b *testing.B) {
 	require.NoError(b, op.SetOutputs([]operator.Operator{fake}))
 
 	go func() {
-		for {
-			<-fake.Received
+		for range fake.Received { //nolint:revive
+			// Nothing to do
 		}
 	}()
 
@@ -818,6 +818,10 @@ func BenchmarkRecombine(b *testing.B) {
 		}
 		op.(*Transformer).flushAllSources(ctx)
 	}
+	b.StopTimer()
+
+	require.NoError(b, op.Stop())
+	close(fake.Received)
 }
 
 func BenchmarkRecombineLimitTrigger(b *testing.B) {
@@ -835,8 +839,8 @@ func BenchmarkRecombineLimitTrigger(b *testing.B) {
 	require.NoError(b, op.Start(nil))
 
 	go func() {
-		for {
-			<-fake.Received
+		for range fake.Received { //nolint:revive
+			// Nothing to do
 		}
 	}()
 
@@ -855,6 +859,10 @@ func BenchmarkRecombineLimitTrigger(b *testing.B) {
 		require.NoError(b, op.ProcessBatch(ctx, []*entry.Entry{start, next}))
 		op.(*Transformer).flushAllSources(ctx)
 	}
+	b.StopTimer()
+
+	require.NoError(b, op.Stop())
+	close(fake.Received)
 }
 
 func TestTimeout(t *testing.T) {

Original file line number	Diff line number	Diff line change
`@@ -791,8 +791,8 @@ func BenchmarkRecombine(b *testing.B) {`
`791`	`791`	`require.NoError(b, op.SetOutputs([]operator.Operator{fake}))`
`792`	`792`
`793`	`793`	`go func() {`
`794`		`- for {`
`795`		`- <-fake.Received`
	`794`	`+ for range fake.Received { //nolint:revive`
	`795`	`+ // Nothing to do`
`796`	`796`	`}`
`797`	`797`	`}()`
`798`	`798`
`@@ -818,6 +818,10 @@ func BenchmarkRecombine(b *testing.B) {`
`818`	`818`	`}`
`819`	`819`	`op.(*Transformer).flushAllSources(ctx)`
`820`	`820`	`}`
	`821`	`+ b.StopTimer()`
	`822`	`+`
	`823`	`+ require.NoError(b, op.Stop())`
	`824`	`+ close(fake.Received)`
`821`	`825`	`}`
`822`	`826`
`823`	`827`	`func BenchmarkRecombineLimitTrigger(b *testing.B) {`
`@@ -835,8 +839,8 @@ func BenchmarkRecombineLimitTrigger(b *testing.B) {`
`835`	`839`	`require.NoError(b, op.Start(nil))`
`836`	`840`
`837`	`841`	`go func() {`
`838`		`- for {`
`839`		`- <-fake.Received`
	`842`	`+ for range fake.Received { //nolint:revive`
	`843`	`+ // Nothing to do`
`840`	`844`	`}`
`841`	`845`	`}()`
`842`	`846`
`@@ -855,6 +859,10 @@ func BenchmarkRecombineLimitTrigger(b *testing.B) {`
`855`	`859`	`require.NoError(b, op.ProcessBatch(ctx, []*entry.Entry{start, next}))`
`856`	`860`	`op.(*Transformer).flushAllSources(ctx)`
`857`	`861`	`}`
	`862`	`+ b.StopTimer()`
	`863`	`+`
	`864`	`+ require.NoError(b, op.Stop())`
	`865`	`+ close(fake.Received)`
`858`	`866`	`}`
`859`	`867`
`860`	`868`	`func TestTimeout(t *testing.T) {`