This is an automated email from the ASF dual-hosted git repository.
wilfred-s 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 b92cd121 [YUNIKORN-3364] Fix event stream history replay blocking
forever on disconnect (#1133)
b92cd121 is described below
commit b92cd1215feee0fd3319600ec46a6e219605580d
Author: PoiBlackTea <[email protected]>
AuthorDate: Wed Sep 2 14:09:16 2026 +0200
[YUNIKORN-3364] Fix event stream history replay blocking forever on
disconnect (#1133)
EventStreaming.CreateEventStream replayed historical events with a bare
blocking send (consumer <- event) outside the forwarder's select loop.
When a client requested non-zero history and disconnected before the stream
opened, or when history events exceeded channel buffer size (1000), the send
blocked indefinitely. Since the goroutine had not yet entered the select
loop,
neither stop channel (stop or e.stopCh) could be reached, causing goroutine
and memory leaks.
Wrap the history-replay send in a select block with e.stopCh and stop cases,
and add a unit test to verify early disconnect during history replay.
Closes: #1133
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/events/event_streaming.go | 10 +++++++++-
pkg/events/event_streaming_test.go | 15 +++++++++++++++
2 files changed, 24 insertions(+), 1 deletion(-)
diff --git a/pkg/events/event_streaming.go b/pkg/events/event_streaming.go
index 59daac50..0aabef31 100644
--- a/pkg/events/event_streaming.go
+++ b/pkg/events/event_streaming.go
@@ -106,7 +106,15 @@ func (e *EventStreaming) CreateEventStream(name string,
count uint64) *EventStre
// map, so "local" will also contain the new event.
seen := make(map[*si.EventRecord]bool)
for _, event := range history {
- consumer <- event
+ select {
+ case consumer <- event:
+ case <-e.stopCh:
+ close(consumer)
+ return
+ case <-stop:
+ close(consumer)
+ return
+ }
seen[event] = true
}
for {
diff --git a/pkg/events/event_streaming_test.go
b/pkg/events/event_streaming_test.go
index 498f540e..dcb5af75 100644
--- a/pkg/events/event_streaming_test.go
+++ b/pkg/events/event_streaming_test.go
@@ -162,6 +162,21 @@ func TestGetEventStreams(t *testing.T) {
assert.Equal(t, 0, len(streaming.eventStreams))
}
+func TestEventStreaming_HistoryReplay_SlowConsumer(t *testing.T) {
+ buffer := newEventRingBuffer(1500)
+ for i := 0; i < 1500; i++ {
+ buffer.Add(&si.EventRecord{TimestampNano: int64(i)})
+ }
+ streaming := NewEventStreaming(buffer)
+ defer streaming.Close()
+
+ // requesting more historical events than defaultChannelBufSize (1000)
and disconnecting
+ es := streaming.CreateEventStream("test", 1500)
+ streaming.RemoveEventStream(es)
+
+ assert.Equal(t, 0, len(streaming.eventStreams))
+}
+
func receive(t *testing.T, input <-chan *si.EventRecord) *si.EventRecord {
select {
case event := <-input:
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
