This is an automated email from the ASF dual-hosted git repository.
manirajv06 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new 41ae1cc5 [YUNIKORN-3372] Clean up event streams in tests to prevent
goroutine leaks (#1135)
41ae1cc5 is described below
commit 41ae1cc5c9cd2dea6ea6defb3e930c740bae1828
Author: PoiBlackTea <[email protected]>
AuthorDate: Fri Aug 28 12:30:27 2026 +0530
[YUNIKORN-3372] Clean up event streams in tests to prevent goroutine leaks
(#1135)
Several tests created event streams via CreateEventStream without calling
RemoveStream/RemoveEventStream, leaving the forwarder goroutine running
after
the test finished.
Add proper cleanup calls (RemoveEventStream / RemoveStream) across all
event-streaming unit tests to prevent goroutine and channel leaks.
Closes: #1135
Signed-off-by: mani <[email protected]>
---
pkg/events/event_streaming_test.go | 12 ++++++++++--
pkg/events/event_system_test.go | 3 ++-
2 files changed, 12 insertions(+), 3 deletions(-)
diff --git a/pkg/events/event_streaming_test.go
b/pkg/events/event_streaming_test.go
index c559c424..498f540e 100644
--- a/pkg/events/event_streaming_test.go
+++ b/pkg/events/event_streaming_test.go
@@ -97,6 +97,8 @@ func TestEventStreaming_WithHistoryCount(t *testing.T) {
assert.Equal(t, int64(6), received1.TimestampNano)
assert.Equal(t, int64(9), received2.TimestampNano)
assert.Equal(t, int64(10), received3.TimestampNano)
+ streaming.RemoveEventStream(es)
+ assert.Equal(t, 0, len(streaming.eventStreams))
}
func TestEventStreaming_TwoConsumers(t *testing.T) {
@@ -118,6 +120,9 @@ func TestEventStreaming_TwoConsumers(t *testing.T) {
assert.Equal(t, 0, len(streaming.eventStreams[es1].consumer))
assert.Equal(t, 0, len(streaming.eventStreams[es2].local))
assert.Equal(t, 0, len(streaming.eventStreams[es2].consumer))
+ streaming.RemoveEventStream(es1)
+ streaming.RemoveEventStream(es2)
+ assert.Equal(t, 0, len(streaming.eventStreams))
}
func TestEventStreaming_SlowConsumer(t *testing.T) {
@@ -139,12 +144,12 @@ func TestGetEventStreams(t *testing.T) {
streaming := NewEventStreaming(buffer)
defer streaming.Close()
- streaming.CreateEventStream("test-1", 0)
+ es1 := streaming.CreateEventStream("test-1", 0)
streams := streaming.GetEventStreams()
assert.Equal(t, 1, len(streams))
assert.Equal(t, "test-1", streams[0].Name)
- streaming.CreateEventStream("test-2", 0)
+ es2 := streaming.CreateEventStream("test-2", 0)
streams = streaming.GetEventStreams()
assert.Equal(t, 2, len(streams))
names := make(map[string]bool)
@@ -152,6 +157,9 @@ func TestGetEventStreams(t *testing.T) {
names[streams[1].Name] = true
assert.Assert(t, names["test-2"])
assert.Assert(t, names["test-1"])
+ streaming.RemoveEventStream(es1)
+ streaming.RemoveEventStream(es2)
+ assert.Equal(t, 0, len(streaming.eventStreams))
}
func receive(t *testing.T, input <-chan *si.EventRecord) *si.EventRecord {
diff --git a/pkg/events/event_system_test.go b/pkg/events/event_system_test.go
index ff3d1320..460b011e 100644
--- a/pkg/events/event_system_test.go
+++ b/pkg/events/event_system_test.go
@@ -165,7 +165,8 @@ func TestEventStreaming(t *testing.T) {
eventSystem.StartService()
defer eventSystem.Stop()
- eventSystem.CreateEventStream("test", 10)
+ stream := eventSystem.CreateEventStream("test", 10)
+ defer eventSystem.RemoveStream(stream)
streams := eventSystem.GetEventStreams()
assert.Equal(t, 1, len(streams))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
