Skip to content
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Commit fbee37a

Browse files
akarnokdakarnokd
authoredApr 27, 2020
3.x: Fix scheduled tasks' fatal exception behavior (#6956)
* 3.x: Fix scheduled tasks' fatal exception behavior * Fix direct periodic tasks not stopping upon crash. * Fix the mistake introduced in the previous commit. * Ensure task exception is rethrown so that the parent FutureTask can end * Update the abstract Scheduler's tasks too * Adjust some test expectation with DisposeTask
1 parent b6a994f commit fbee37a

15 files changed

+425
-36
lines changed
 

‎src/main/java/io/reactivex/rxjava3/core/Scheduler.java

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,9 @@
1818

1919
import io.reactivex.rxjava3.annotations.*;
2020
import io.reactivex.rxjava3.disposables.Disposable;
21-
import io.reactivex.rxjava3.exceptions.Exceptions;
2221
import io.reactivex.rxjava3.functions.Function;
2322
import io.reactivex.rxjava3.internal.disposables.*;
2423
import io.reactivex.rxjava3.internal.schedulers.*;
25-
import io.reactivex.rxjava3.internal.util.ExceptionHelper;
2624
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
2725
import io.reactivex.rxjava3.schedulers.SchedulerRunnableIntrospection;
2826

@@ -542,9 +540,10 @@ public void run() {
542540
try {
543541
run.run();
544542
} catch (Throwable ex) {
545-
Exceptions.throwIfFatal(ex);
546-
worker.dispose();
547-
throw ExceptionHelper.wrapOrThrow(ex);
543+
// Exceptions.throwIfFatal(ex); nowhere to go
544+
dispose();
545+
RxJavaPlugins.onError(ex);
546+
throw ex;
548547
}
549548
}
550549
}
@@ -586,7 +585,13 @@ static final class DisposeTask implements Disposable, Runnable, SchedulerRunnabl
586585
public void run() {
587586
runner = Thread.currentThread();
588587
try {
589-
decoratedRun.run();
588+
try {
589+
decoratedRun.run();
590+
} catch (Throwable ex) {
591+
// Exceptions.throwIfFatal(e); nowhere to go
592+
RxJavaPlugins.onError(ex);
593+
throw ex;
594+
}
590595
} finally {
591596
dispose();
592597
runner = null;

‎src/main/java/io/reactivex/rxjava3/internal/schedulers/ExecutorScheduler.java

Lines changed: 22 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -320,6 +320,10 @@ public void run() {
320320
}
321321
try {
322322
actual.run();
323+
} catch (Throwable ex) {
324+
// Exceptions.throwIfFatal(ex); nowhere to go
325+
RxJavaPlugins.onError(ex);
326+
throw ex;
323327
} finally {
324328
lazySet(true);
325329
}
@@ -386,7 +390,13 @@ public void run() {
386390
thread = Thread.currentThread();
387391
if (compareAndSet(READY, RUNNING)) {
388392
try {
389-
run.run();
393+
try {
394+
run.run();
395+
} catch (Throwable ex) {
396+
// Exceptions.throwIfFatal(ex); nowhere to go
397+
RxJavaPlugins.onError(ex);
398+
throw ex;
399+
}
390400
} finally {
391401
thread = null;
392402
if (compareAndSet(RUNNING, FINISHED)) {
@@ -463,11 +473,17 @@ public void run() {
463473
Runnable r = get();
464474
if (r != null) {
465475
try {
466-
r.run();
467-
} finally {
468-
lazySet(null);
469-
timed.lazySet(DisposableHelper.DISPOSED);
470-
direct.lazySet(DisposableHelper.DISPOSED);
476+
try {
477+
r.run();
478+
} finally {
479+
lazySet(null);
480+
timed.lazySet(DisposableHelper.DISPOSED);
481+
direct.lazySet(DisposableHelper.DISPOSED);
482+
}
483+
} catch (Throwable ex) {
484+
// Exceptions.throwIfFatal(ex); nowhere to go
485+
RxJavaPlugins.onError(ex);
486+
throw ex;
471487
}
472488
}
473489
}

‎src/main/java/io/reactivex/rxjava3/internal/schedulers/InstantPeriodicTask.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
import java.util.concurrent.atomic.AtomicReference;
2121

2222
import io.reactivex.rxjava3.disposables.Disposable;
23-
import io.reactivex.rxjava3.exceptions.Exceptions;
2423
import io.reactivex.rxjava3.internal.functions.Functions;
2524
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
2625

@@ -54,12 +53,13 @@ public Void call() {
5453
runner = Thread.currentThread();
5554
try {
5655
task.run();
57-
setRest(executor.submit(this));
5856
runner = null;
57+
setRest(executor.submit(this));
5958
} catch (Throwable ex) {
60-
Exceptions.throwIfFatal(ex);
59+
// Exceptions.throwIfFatal(ex); nowhere to go
6160
runner = null;
6261
RxJavaPlugins.onError(ex);
62+
throw ex;
6363
}
6464
return null;
6565
}

‎src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTask.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@
1616

1717
package io.reactivex.rxjava3.internal.schedulers;
1818

19-
import io.reactivex.rxjava3.exceptions.Exceptions;
2019
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
2120

2221
/**
@@ -39,10 +38,11 @@ public void run() {
3938
runnable.run();
4039
runner = null;
4140
} catch (Throwable ex) {
42-
Exceptions.throwIfFatal(ex);
41+
// Exceptions.throwIfFatal(ex); nowhere to go
4342
runner = null;
44-
lazySet(FINISHED);
43+
dispose();
4544
RxJavaPlugins.onError(ex);
45+
throw ex;
4646
}
4747
}
4848
}

‎src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectTask.java

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818

1919
import java.util.concurrent.Callable;
2020

21+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
22+
2123
/**
2224
* A Callable to be submitted to an ExecutorService that runs a Runnable
2325
* action and manages completion/cancellation.
@@ -35,10 +37,16 @@ public ScheduledDirectTask(Runnable runnable) {
3537
public Void call() {
3638
runner = Thread.currentThread();
3739
try {
38-
runnable.run();
39-
} finally {
40-
lazySet(FINISHED);
41-
runner = null;
40+
try {
41+
runnable.run();
42+
} finally {
43+
lazySet(FINISHED);
44+
runner = null;
45+
}
46+
} catch (Throwable ex) {
47+
// Exceptions.throwIfFatal(e); nowhere to go
48+
RxJavaPlugins.onError(ex);
49+
throw ex;
4250
}
4351
return null;
4452
}

‎src/main/java/io/reactivex/rxjava3/internal/schedulers/ScheduledRunnable.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ public void run() {
6666
} catch (Throwable e) {
6767
// Exceptions.throwIfFatal(e); nowhere to go
6868
RxJavaPlugins.onError(e);
69+
throw e;
6970
}
7071
} finally {
7172
lazySet(THREAD_INDEX, null);

‎src/test/java/io/reactivex/rxjava3/core/DisposeTaskTest.java

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/**
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava3.core;
15+
16+
import static org.junit.Assert.fail;
17+
import static org.testng.Assert.assertTrue;
18+
19+
import org.junit.Test;
20+
21+
import io.reactivex.rxjava3.core.Scheduler.DisposeTask;
22+
import io.reactivex.rxjava3.exceptions.TestException;
23+
import io.reactivex.rxjava3.schedulers.Schedulers;
24+
import io.reactivex.rxjava3.testsupport.TestHelper;
25+
26+
public class DisposeTaskTest extends RxJavaTest {
27+
28+
@Test
29+
public void runnableThrows() throws Throwable {
30+
TestHelper.withErrorTracking(errors -> {
31+
32+
Scheduler.Worker worker = Schedulers.single().createWorker();
33+
34+
DisposeTask task = new DisposeTask(() -> {
35+
throw new TestException();
36+
}, worker);
37+
38+
try {
39+
task.run();
40+
fail("Should have thrown!");
41+
} catch (TestException expected) {
42+
// expected
43+
}
44+
45+
TestHelper.assertUndeliverable(errors, 0, TestException.class);
46+
47+
assertTrue(worker.isDisposed());
48+
});
49+
}
50+
}

‎src/test/java/io/reactivex/rxjava3/core/PeriodicDirectTaskTest.java

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
/**
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava3.core;
15+
16+
import static org.junit.Assert.fail;
17+
import static org.testng.Assert.assertTrue;
18+
19+
import java.util.List;
20+
21+
import org.junit.Test;
22+
23+
import io.reactivex.rxjava3.core.Scheduler.PeriodicDirectTask;
24+
import io.reactivex.rxjava3.exceptions.TestException;
25+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
26+
import io.reactivex.rxjava3.schedulers.Schedulers;
27+
import io.reactivex.rxjava3.testsupport.TestHelper;
28+
29+
public class PeriodicDirectTaskTest extends RxJavaTest {
30+
31+
@Test
32+
public void runnableThrows() {
33+
List<Throwable> errors = TestHelper.trackPluginErrors();
34+
try {
35+
Scheduler.Worker worker = Schedulers.single().createWorker();
36+
37+
PeriodicDirectTask task = new PeriodicDirectTask(() -> {
38+
throw new TestException();
39+
}, worker);
40+
41+
try {
42+
task.run();
43+
fail("Should have thrown!");
44+
} catch (TestException expected) {
45+
// expected
46+
}
47+
48+
TestHelper.assertUndeliverable(errors, 0, TestException.class);
49+
50+
assertTrue(worker.isDisposed());
51+
52+
task.run();
53+
} finally {
54+
RxJavaPlugins.reset();
55+
}
56+
}
57+
}

‎src/test/java/io/reactivex/rxjava3/internal/schedulers/BooleanRunnableTest.java

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/**
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava3.internal.schedulers;
15+
16+
import static org.junit.Assert.fail;
17+
18+
import java.util.List;
19+
20+
import org.junit.Test;
21+
22+
import io.reactivex.rxjava3.core.RxJavaTest;
23+
import io.reactivex.rxjava3.exceptions.TestException;
24+
import io.reactivex.rxjava3.internal.schedulers.ExecutorScheduler.ExecutorWorker.BooleanRunnable;
25+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
26+
import io.reactivex.rxjava3.testsupport.TestHelper;
27+
28+
public class BooleanRunnableTest extends RxJavaTest {
29+
30+
@Test
31+
public void runnableThrows() {
32+
List<Throwable> errors = TestHelper.trackPluginErrors();
33+
try {
34+
BooleanRunnable task = new BooleanRunnable(() -> {
35+
throw new TestException();
36+
});
37+
38+
try {
39+
task.run();
40+
fail("Should have thrown!");
41+
} catch (TestException expected) {
42+
// expected
43+
}
44+
45+
TestHelper.assertUndeliverable(errors, 0, TestException.class);
46+
} finally {
47+
RxJavaPlugins.reset();
48+
}
49+
}
50+
}

‎src/test/java/io/reactivex/rxjava3/internal/schedulers/InstantPeriodicTaskTest.java

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,12 @@ public void run() {
4444
}
4545
}, exec);
4646

47-
assertNull(task.call());
47+
try {
48+
task.call();
49+
fail("Should have thrown!");
50+
} catch (TestException excepted) {
51+
// excepted
52+
}
4853

4954
TestHelper.assertUndeliverable(errors, 0, TestException.class);
5055
} finally {

‎src/test/java/io/reactivex/rxjava3/internal/schedulers/InterruptibleRunnableTest.java

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/**
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava3.internal.schedulers;
15+
16+
import static org.junit.Assert.fail;
17+
18+
import java.util.List;
19+
20+
import org.junit.Test;
21+
22+
import io.reactivex.rxjava3.core.RxJavaTest;
23+
import io.reactivex.rxjava3.exceptions.TestException;
24+
import io.reactivex.rxjava3.internal.schedulers.ExecutorScheduler.ExecutorWorker.InterruptibleRunnable;
25+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
26+
import io.reactivex.rxjava3.testsupport.TestHelper;
27+
28+
public class InterruptibleRunnableTest extends RxJavaTest {
29+
30+
@Test
31+
public void runnableThrows() {
32+
List<Throwable> errors = TestHelper.trackPluginErrors();
33+
try {
34+
InterruptibleRunnable task = new InterruptibleRunnable(() -> {
35+
throw new TestException();
36+
}, null);
37+
38+
try {
39+
task.run();
40+
fail("Should have thrown!");
41+
} catch (TestException expected) {
42+
// expected
43+
}
44+
45+
TestHelper.assertUndeliverable(errors, 0, TestException.class);
46+
} finally {
47+
RxJavaPlugins.reset();
48+
}
49+
}
50+
}

‎src/test/java/io/reactivex/rxjava3/internal/schedulers/ScheduledDirectPeriodicTaskTest.java

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@
1313

1414
package io.reactivex.rxjava3.internal.schedulers;
1515

16+
import static org.junit.Assert.fail;
17+
1618
import java.util.List;
1719

1820
import org.junit.Test;
@@ -35,7 +37,12 @@ public void run() {
3537
}
3638
});
3739

38-
task.run();
40+
try {
41+
task.run();
42+
fail("Should have thrown!");
43+
} catch (TestException expected) {
44+
// expected
45+
}
3946

4047
TestHelper.assertUndeliverable(errors, 0, TestException.class);
4148
} finally {

‎src/test/java/io/reactivex/rxjava3/internal/schedulers/ScheduledRunnableTest.java

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -208,7 +208,12 @@ public void run() {
208208
}, set);
209209
set.add(run);
210210

211-
run.run();
211+
try {
212+
run.run();
213+
fail("Should have thrown!");
214+
} catch (TestException expected) {
215+
// expected
216+
}
212217

213218
assertTrue(run.isDisposed());
214219

‎src/test/java/io/reactivex/rxjava3/schedulers/ComputationSchedulerTests.java

Lines changed: 126 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,14 +17,16 @@
1717

1818
import java.util.HashMap;
1919
import java.util.concurrent.*;
20+
import java.util.concurrent.atomic.AtomicInteger;
2021

21-
import io.reactivex.rxjava3.disposables.Disposable;
2222
import org.junit.Test;
2323

2424
import io.reactivex.rxjava3.core.*;
2525
import io.reactivex.rxjava3.core.Scheduler.Worker;
26+
import io.reactivex.rxjava3.disposables.Disposable;
2627
import io.reactivex.rxjava3.functions.*;
2728
import io.reactivex.rxjava3.internal.schedulers.ComputationScheduler;
29+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
2830

2931
public class ComputationSchedulerTests extends AbstractSchedulerConcurrencyTests {
3032

@@ -192,4 +194,127 @@ public void run() {
192194

193195
assertEquals(0, calls[0]);
194196
}
197+
198+
@Test
199+
public void exceptionFromObservableShouldNotBeSwallowed() throws Exception {
200+
CountDownLatch latch = new CountDownLatch(1);
201+
202+
// #3 thread's uncaught exception handler
203+
Scheduler computationScheduler = new ComputationScheduler(new ThreadFactory() {
204+
@Override
205+
public Thread newThread(Runnable r) {
206+
Thread t = new Thread(r);
207+
t.setUncaughtExceptionHandler((thread, throwable) -> {
208+
latch.countDown();
209+
});
210+
return t;
211+
}
212+
});
213+
214+
// #2 RxJava exception handler
215+
RxJavaPlugins.setErrorHandler(h -> {
216+
latch.countDown();
217+
});
218+
219+
// Exceptions, fatal or not, should be handled by
220+
// #1 observer's onError(), or
221+
// #2 RxJava exception handler, or
222+
// #3 thread's uncaught exception handler,
223+
// and should not be swallowed.
224+
try {
225+
226+
// #1 observer's onError()
227+
Observable.create(s -> {
228+
229+
s.onNext(1);
230+
throw new OutOfMemoryError();
231+
})
232+
.subscribeOn(computationScheduler)
233+
.subscribe(v -> { },
234+
e -> { latch.countDown(); }
235+
);
236+
237+
assertTrue(latch.await(2, TimeUnit.SECONDS));
238+
} finally {
239+
RxJavaPlugins.reset();
240+
computationScheduler.shutdown();
241+
}
242+
}
243+
244+
@Test
245+
public void exceptionFromObserverShouldNotBeSwallowed() throws Exception {
246+
CountDownLatch latch = new CountDownLatch(1);
247+
248+
// #3 thread's uncaught exception handler
249+
Scheduler computationScheduler = new ComputationScheduler(new ThreadFactory() {
250+
@Override
251+
public Thread newThread(Runnable r) {
252+
Thread t = new Thread(r);
253+
t.setUncaughtExceptionHandler((thread, throwable) -> {
254+
latch.countDown();
255+
});
256+
return t;
257+
}
258+
});
259+
260+
// #2 RxJava exception handler
261+
RxJavaPlugins.setErrorHandler(h -> {
262+
latch.countDown();
263+
});
264+
265+
// Exceptions, fatal or not, should be handled by
266+
// #1 observer's onError(), or
267+
// #2 RxJava exception handler, or
268+
// #3 thread's uncaught exception handler,
269+
// and should not be swallowed.
270+
try {
271+
272+
// #1 observer's onError()
273+
Flowable.interval(500, TimeUnit.MILLISECONDS, computationScheduler)
274+
.subscribe(v -> {
275+
throw new OutOfMemoryError();
276+
}, e -> {
277+
latch.countDown();
278+
});
279+
280+
assertTrue(latch.await(2, TimeUnit.SECONDS));
281+
} finally {
282+
RxJavaPlugins.reset();
283+
computationScheduler.shutdown();
284+
}
285+
}
286+
287+
@Test
288+
public void periodicTaskShouldStopOnError() throws Exception {
289+
AtomicInteger repeatCount = new AtomicInteger();
290+
291+
Schedulers.computation().schedulePeriodicallyDirect(new Runnable() {
292+
@Override
293+
public void run() {
294+
repeatCount.incrementAndGet();
295+
throw new OutOfMemoryError();
296+
}
297+
}, 0, 1, TimeUnit.MILLISECONDS);
298+
299+
Thread.sleep(200);
300+
301+
assertEquals(1, repeatCount.get());
302+
}
303+
304+
@Test
305+
public void periodicTaskShouldStopOnError2() throws Exception {
306+
AtomicInteger repeatCount = new AtomicInteger();
307+
308+
Schedulers.computation().schedulePeriodicallyDirect(new Runnable() {
309+
@Override
310+
public void run() {
311+
repeatCount.incrementAndGet();
312+
throw new OutOfMemoryError();
313+
}
314+
}, 0, 1, TimeUnit.NANOSECONDS);
315+
316+
Thread.sleep(200);
317+
318+
assertEquals(1, repeatCount.get());
319+
}
195320
}

‎src/test/java/io/reactivex/rxjava3/schedulers/SchedulerTest.java

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -61,18 +61,28 @@ public void run() {
6161
assertEquals(2, count[0]);
6262
}
6363

64-
@Test(expected = TestException.class)
65-
public void periodicDirectThrows() {
66-
TestScheduler scheduler = new TestScheduler();
64+
@Test
65+
public void periodicDirectThrows() throws Throwable {
66+
TestHelper.withErrorTracking(errors -> {
67+
TestScheduler scheduler = new TestScheduler();
6768

68-
scheduler.schedulePeriodicallyDirect(new Runnable() {
69-
@Override
70-
public void run() {
71-
throw new TestException();
69+
try {
70+
scheduler.schedulePeriodicallyDirect(new Runnable() {
71+
@Override
72+
public void run() {
73+
throw new TestException();
74+
}
75+
}, 100, 100, TimeUnit.MILLISECONDS);
76+
77+
scheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS);
78+
79+
fail("Should have thrown!");
80+
} catch (TestException expected) {
81+
// expected
7282
}
73-
}, 100, 100, TimeUnit.MILLISECONDS);
7483

75-
scheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS);
84+
TestHelper.assertUndeliverable(errors, 0, TestException.class);
85+
});
7686
}
7787

7888
@Test
@@ -233,7 +243,7 @@ public void run() {
233243

234244
Thread.sleep(250);
235245

236-
assertEquals(1, list.size());
246+
assertTrue(list.size() >= 1);
237247
TestHelper.assertUndeliverable(list, 0, TestException.class, null);
238248

239249
} finally {

0 commit comments

Comments
 (0)
Please sign in to comment.