Skip to content

Fix DispatchStreamProcessorTest.raceToStream flakiness #8275

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

Merged
akarnokd merged 3 commits into from
Aug 6, 2026
+107 −56
Merged

Fix DispatchStreamProcessorTest.raceToStream flakiness #8275

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0051aee
Initial plan
Copilot Aug 6, 2026
c9cc62e
Fix raceToStream flakiness in DispatchStreamProcessorTest
Copilot Aug 6, 2026
7b65a64
Polish awaitCondition to use TimeUnit wall-clock deadline
Copilot Aug 6, 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
8 changes: 7 additions & 1 deletion src/main/java/io/reactivex/rxjava4/processors/DispatchStreamProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,14 @@ public final class DispatchStreamProcessor<T> implements StreamProcessor<T, T> {
@Override
public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) {
var result = new DispatchStreamer<T>(this);
cancellation.add(result);
if (add(result)) {
// Register with cancellation after a successful add. If cancellation
// already disposed the streamer (or dispose races with add), remove again
// because the first remove may have observed the pre-add array.
cancellation.add(result);
if (result.isDisposed()) {
remove(result);
}
return result;
}
var t = terminalEvent;
Expand Down
52 changes: 23 additions & 29 deletions .../java/io/reactivex/rxjava4/internal/operators/streamable/DispatchStreamProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public void normal() throws Throwable {

ts.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000);
assertHasStreamers(dsp, 1000);

for (int i = 1; i < 6; i++) {
dsp.next(i).toCompletableFuture().join();
Expand All @@ -50,7 +50,7 @@ public void normal() throws Throwable {
ts.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3, 4, 5);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertTrue(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand All @@ -69,7 +69,7 @@ public void endsInError() throws Throwable {

ts.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000);
assertHasStreamers(dsp, 1000);

for (int i = 1; i < 6; i++) {
dsp.next(i).toCompletableFuture().join();
Expand All @@ -80,7 +80,7 @@ public void endsInError() throws Throwable {
ts.awaitDone(5, TimeUnit.SECONDS)
.assertFailure(TestException.class, 1, 2, 3, 4, 5);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertFalse(dsp.hasComplete(), "dsp has completed?");
assertTrue(dsp.hasThrowable(), "dsp has no throwable?");
assertSame(te, dsp.getThrowable(), "dsp has the wrong throwable?");
Expand All @@ -100,9 +100,7 @@ public void normalDebug() throws Throwable {

ts.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000);

assertTrue(dsp.hasStreamers(), "dsp has no streamers?");
assertHasStreamers(dsp, 1000);

for (int i = 1; i < 6; i++) {
dsp.next(i).toCompletableFuture().join();
Expand All @@ -112,7 +110,7 @@ public void normalDebug() throws Throwable {
ts.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3, 4, 5);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertTrue(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand Down Expand Up @@ -178,16 +176,14 @@ public void normalTake3AltDebug() throws Throwable {

ts.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000);
assertHasStreamers(dsp, 1000);

for (int i = 1; i < 4; i++) {
IO.println(i + " -> next");
dsp.next(i).toCompletableFuture().join();
}

awaitNoStreamers(dsp, 1000);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);

for (int i = 4; i < 6; i++) {
IO.println(i + " -> next");
Expand All @@ -201,7 +197,7 @@ public void normalTake3AltDebug() throws Throwable {
ts.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertTrue(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand All @@ -222,7 +218,7 @@ public void normalTake3Debug() throws Throwable {

ts.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000);
assertHasStreamers(dsp, 1000);

for (int i = 1; i < 6; i++) {
dsp.next(i).toCompletableFuture().join();
Expand All @@ -235,7 +231,7 @@ public void normalTake3Debug() throws Throwable {
ts.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertTrue(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand All @@ -257,7 +253,7 @@ public void normalMulti() throws Throwable {
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
ts2.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000, 2);
assertHasStreamers(dsp, 1000, 2);

for (int i = 1; i < 6; i++) {
dsp.next(i).toCompletableFuture().join();
Expand All @@ -270,7 +266,7 @@ public void normalMulti() throws Throwable {
ts2.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3, 4, 5);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertTrue(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand All @@ -291,7 +287,7 @@ public void normalMultiOtherCancels() throws Throwable {
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
ts2.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000, 2);
assertHasStreamers(dsp, 1000, 2);

ts2.cancel();

Expand All @@ -305,7 +301,7 @@ public void normalMultiOtherCancels() throws Throwable {

ts2.assertEmpty();

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertTrue(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand All @@ -326,7 +322,7 @@ public void normalMultiFirstCancels() throws Throwable {
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
ts2.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000, 2);
assertHasStreamers(dsp, 1000, 2);

ts.cancel();

Expand All @@ -340,7 +336,7 @@ public void normalMultiFirstCancels() throws Throwable {

ts.assertEmpty();

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertTrue(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand All @@ -367,14 +363,14 @@ public void raceToStream() throws Throwable {
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
ts2.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000);
// Both subscribers must be fully attached before cancellation,
// otherwise a late stream() can re-add a streamer after the wait.
assertHasStreamers(dsp, 1000, 2);

ts.cancel();
ts2.cancel();

awaitNoStreamers(dsp, 1000);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertFalse(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand All @@ -394,13 +390,11 @@ public void comeAndGo() throws Throwable {

ts.awaitOnSubscribe(1, TimeUnit.SECONDS);

awaitStreamers(dsp, 1000);
assertHasStreamers(dsp, 1000);

ts.cancel();

awaitNoStreamers(dsp, 1000);

assertFalse(dsp.hasStreamers(), "dsp has streamers?");
assertNoStreamers(dsp, 1000);
assertFalse(dsp.hasComplete(), "dsp has completed?");
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
Expand Down
103 changes: 77 additions & 26 deletions src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@

import java.lang.ref.Cleaner;
import java.util.*;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.*;
import java.util.function.*;

import org.junit.jupiter.api.*;
Expand Down Expand Up @@ -119,13 +119,7 @@ public static <T> StreamableInterceptConfig<T> debugIntercept() {
public static void awaitStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
throws InterruptedException, TimeoutException
{
long timeout = timeoutMillis * 1_000_000L;
while (!sp.hasStreamers()) {
Thread.sleep(0, 1000);
if (--timeout <= 0L) {
throw new TimeoutException("hasStreamers still false");
}
}
awaitCondition(true, sp::hasStreamers, timeoutMillis, "hasStreamers still false");
}

/**
Expand All @@ -140,13 +134,8 @@ public static void awaitStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
public static void awaitStreamers(StreamProcessor<?, ?> sp, long timeoutMillis, int atLeast)
throws InterruptedException, TimeoutException
{
long timeout = timeoutMillis * 1_000_000L;
while (sp.streamerCount() < atLeast) {
Thread.sleep(0, 1000);
if (--timeout <= 0L) {
throw new TimeoutException("hasStreamers still false");
}
}
awaitCondition(true, () -> sp.streamerCount() >= atLeast, timeoutMillis,
"streamerCount still below " + atLeast + " (was " + sp.streamerCount() + ")");
}

/**
Expand All @@ -160,12 +149,55 @@ public static void awaitStreamers(StreamProcessor<?, ?> sp, long timeoutMillis,
public static void awaitNoStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
throws InterruptedException, TimeoutException
{
long timeout = timeoutMillis * 1_000_000L;
while (sp.hasStreamers()) {
Thread.sleep(0, 1000);
if (--timeout <= 0L) {
throw new TimeoutException("hasStreamers still false");
}
awaitCondition(false, sp::hasStreamers, timeoutMillis, "hasStreamers still true");
}

/**
* Awaits until the processor has no streamers and asserts that state.
* @param sp the processor
* @param timeoutMillis how long to wait for the streamer(s) to leave
* @throws InterruptedException if the sleep is interrupted
* @throws TimeoutException if the wait times out
*/
public static void assertNoStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
throws InterruptedException, TimeoutException
{
awaitNoStreamers(sp, timeoutMillis);
if (sp.hasStreamers()) {
throw new AssertionError("Processor still has streamers: " + sp.streamerCount());
}
}

/**
* Awaits until the processor has at least one streamer and asserts that state.
* @param sp the processor
* @param timeoutMillis how long to wait for the streamer(s) to arrive
* @throws InterruptedException if the sleep is interrupted
* @throws TimeoutException if the wait times out
*/
public static void assertHasStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
throws InterruptedException, TimeoutException
{
awaitStreamers(sp, timeoutMillis);
if (!sp.hasStreamers()) {
throw new AssertionError("Processor has no streamers");
}
}

/**
* Awaits until the processor has at least {@code atLeast} streamers and asserts that state.
* @param sp the processor
* @param timeoutMillis how long to wait for the streamer(s) to arrive
* @param atLeast the minimum number of streamers expected
* @throws InterruptedException if the sleep is interrupted
* @throws TimeoutException if the wait times out
*/
public static void assertHasStreamers(StreamProcessor<?, ?> sp, long timeoutMillis, int atLeast)
throws InterruptedException, TimeoutException
{
awaitStreamers(sp, timeoutMillis, atLeast);
if (sp.streamerCount() < atLeast) {
throw new AssertionError("Processor streamerCount below " + atLeast + ": " + sp.streamerCount());
}
}

Expand All @@ -181,12 +213,31 @@ public static void awaitNoStreamers(StreamProcessor<?, ?> sp, long timeoutMillis
*/
public static void awaitCondition(boolean value, @NonNull BooleanSupplier condition, long timeoutMillis)
throws InterruptedException, TimeoutException {
long timeout = timeoutMillis * 1_000_000L;
while (condition.getAsBoolean() != value) {
Thread.sleep(0, 1000);
if (--timeout <= 0L) {
throw new TimeoutException("condition still " + (!value));
awaitCondition(value, condition, timeoutMillis, "condition still " + (!value));
}

/**
* Awaits a given {@link BooleanSupplier} to return the expected {@code value}
* within the given wall-clock time period.
* @param value the expected value within the timeout period
* @param condition the condition to repeatedly call to assess the state
* @param timeoutMillis how long to wait for the condition to become as expected
* @param timeoutMessage the message used when the wait times out
* @throws InterruptedException if the sleep is interrupted
* @throws TimeoutException if the wait times out
*/
public static void awaitCondition(boolean value, @NonNull BooleanSupplier condition, long timeoutMillis,
@NonNull String timeoutMessage)
throws InterruptedException, TimeoutException {
long end = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
for (;;) {
if (condition.getAsBoolean() == value) {
return;
}
if (System.nanoTime() >= end) {
throw new TimeoutException(timeoutMessage);
}
Thread.sleep(0, 1000);
}
}
}