Skip to content

try to fix intermittent test failure in HubSpec #2943

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
pjfanning wants to merge 7 commits into
base: main
Choose a base branch
from
Draft

try to fix intermittent test failure in HubSpec #2943

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6960b1e
BroadcastHub: serialize unregister before re-register
Copilot May 7, 2026
0b9df14
Tighten BroadcastHub handoff race handling
Copilot May 7, 2026
af7b1e0
Use deterministic BroadcastHub handoff regression
Copilot May 7, 2026
8f8f49c
Revert BroadcastHub production changes
Copilot May 7, 2026
5e473f6
Keep BroadcastHub fix test-only
Copilot May 7, 2026
a9102c6
Update HubSpec.scala
pjfanning May 7, 2026
f11feee
Rewrite HubSpec sequential consumer test to eliminate race condition
Copilot May 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/HubSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -402,8 +402,18 @@ class HubSpec extends StreamSpec {

"ensure that subsequent consumers see subsequent elements without gap" in {
val source = Source(1 to 20).runWith(BroadcastHub.sink(8))

// Let the first consumer take exactly 10 elements and complete naturally.
// GraphStageLogic.internalCompleteStage cancels inlets before completing outlets,
// so the hub's UnRegister event is enqueued in the hub actor's mailbox before
// Sink.seq's postStop resolves the future. The second consumer's RegistrationPending
// is only sent after futureValue returns, guaranteeing UnRegister is processed first.
source.take(10).runWith(Sink.seq).futureValue should ===(1 to 10)
source.take(10).runWith(Sink.seq).futureValue should ===(11 to 20)

val secondConsumer = source.runWith(TestSink[Int]())
secondConsumer.request(10)
secondConsumer.expectNextN((11 to 20).toVector)
secondConsumer.expectComplete()
}

"send the same elements to consumers of different speed attaching around the same time" in {
Expand Down
Loading