Skip to content

Commit bdf13a1

Browse files
benjamintboylebenjamintboyle
authored
Fix intermittently failing unit tests (#7213)
* Fix intermittently failing tests, mostly increasing timeout Fix several unit tests that are intermittently failing. All fixes involve increasing timeouts. Adjusted formatting in several areas within updated tests. * Remove formatting/refactoring from previous commit Superfluous formatting and refactoring was making review impossible.
1 parent e600b53 commit bdf13a1

File tree

7 files changed

+51
-37
lines changed

7 files changed

+51
-37
lines changed

‎src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatMapSchedulerTest.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -355,7 +355,7 @@ public Flowable<String> apply(Integer t) throws Throwable {
355355
}
356356

357357
@Test
358-
public void issue2890NoStackoverflow() throws InterruptedException {
358+
public void issue2890NoStackoverflow() throws InterruptedException, TimeoutException {
359359
final ExecutorService executor = Executors.newFixedThreadPool(2);
360360
final Scheduler sch = Schedulers.from(executor);
361361

@@ -400,7 +400,11 @@ public void onError(Throwable e) {
400400
}
401401
});
402402

403-
executor.awaitTermination(20000, TimeUnit.MILLISECONDS);
403+
long awaitTerminationTimeoutMillis = 100_000;
404+
if (!executor.awaitTermination(awaitTerminationTimeoutMillis, TimeUnit.MILLISECONDS)) {
405+
throw new TimeoutException("Completed " + counter.get() + "/" + n + " before timed out after "
406+
+ awaitTerminationTimeoutMillis + " milliseconds.");
407+
}
404408

405409
assertEquals(n, counter.get());
406410
}

‎src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableConcatTest.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -729,7 +729,7 @@ public void subscribe(Subscriber<? super String> s) {
729729
}
730730

731731
@Test
732-
public void issue2890NoStackoverflow() throws InterruptedException {
732+
public void issue2890NoStackoverflow() throws InterruptedException, TimeoutException {
733733
final ExecutorService executor = Executors.newFixedThreadPool(2);
734734
final Scheduler sch = Schedulers.from(executor);
735735

@@ -774,7 +774,11 @@ public void onError(Throwable e) {
774774
}
775775
});
776776

777-
executor.awaitTermination(20000, TimeUnit.MILLISECONDS);
777+
long awaitTerminationTimeoutMillis = 100_000;
778+
if (!executor.awaitTermination(awaitTerminationTimeoutMillis, TimeUnit.MILLISECONDS)) {
779+
throw new TimeoutException("Completed " + counter.get() + "/" + n + " before timed out after "
780+
+ awaitTerminationTimeoutMillis + " milliseconds.");
781+
}
778782

779783
assertEquals(n, counter.get());
780784
}

‎src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableMergeTest.java‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -204,7 +204,7 @@ public void mergeArrayWithThreading() {
204204
TestSubscriber<String> ts = new TestSubscriber<>(stringSubscriber);
205205
m.subscribe(ts);
206206

207-
ts.awaitDone(5, TimeUnit.SECONDS);
207+
ts.awaitDone(10, TimeUnit.SECONDS);
208208
ts.assertNoErrors();
209209

210210
verify(stringSubscriber, never()).onError(any(Throwable.class));
@@ -598,7 +598,7 @@ public void run() {
598598
TestSubscriber<Integer> ts = new TestSubscriber<>();
599599
merge.subscribe(ts);
600600

601-
ts.awaitDone(5, TimeUnit.SECONDS);
601+
ts.awaitDone(10, TimeUnit.SECONDS);
602602
ts.assertComplete();
603603
List<Integer> onNextEvents = ts.values();
604604
assertEquals(300, onNextEvents.size());
@@ -645,7 +645,7 @@ public void run() {
645645
TestSubscriber<Integer> ts = new TestSubscriber<>();
646646
merge.subscribe(ts);
647647

648-
ts.awaitDone(5, TimeUnit.SECONDS);
648+
ts.awaitDone(10, TimeUnit.SECONDS);
649649
ts.assertNoErrors();
650650
ts.assertComplete();
651651
List<Integer> onNextEvents = ts.values();
@@ -706,7 +706,7 @@ public void onNext(Integer t) {
706706
};
707707

708708
Flowable.merge(f1.take(Flowable.bufferSize() * 2), Flowable.just(-99)).subscribe(testSubscriber);
709-
testSubscriber.awaitDone(5, TimeUnit.SECONDS);
709+
testSubscriber.awaitDone(10, TimeUnit.SECONDS);
710710

711711
List<Integer> onNextEvents = testSubscriber.values();
712712

@@ -752,7 +752,7 @@ public void onNext(Integer t) {
752752
};
753753

754754
Flowable.merge(f1.take(Flowable.bufferSize() * 2), f2.take(Flowable.bufferSize() * 2)).observeOn(Schedulers.computation()).subscribe(testSubscriber);
755-
testSubscriber.awaitDone(5, TimeUnit.SECONDS);
755+
testSubscriber.awaitDone(10, TimeUnit.SECONDS);
756756
if (testSubscriber.errors().size() > 0) {
757757
testSubscriber.errors().get(0).printStackTrace();
758758
}
@@ -795,7 +795,7 @@ public void onNext(Integer t) {
795795
};
796796

797797
Flowable.merge(f1).observeOn(Schedulers.computation()).take(Flowable.bufferSize() * 2).subscribe(testSubscriber);
798-
testSubscriber.awaitDone(5, TimeUnit.SECONDS);
798+
testSubscriber.awaitDone(10, TimeUnit.SECONDS);
799799
if (testSubscriber.errors().size() > 0) {
800800
testSubscriber.errors().get(0).printStackTrace();
801801
}
@@ -850,7 +850,7 @@ public void onNext(Integer t) {
850850
};
851851

852852
Flowable.merge(f1).observeOn(Schedulers.computation()).take(Flowable.bufferSize() * 2).subscribe(testSubscriber);
853-
testSubscriber.awaitDone(5, TimeUnit.SECONDS);
853+
testSubscriber.awaitDone(10, TimeUnit.SECONDS);
854854
if (testSubscriber.errors().size() > 0) {
855855
testSubscriber.errors().get(0).printStackTrace();
856856
}
@@ -868,7 +868,7 @@ public void onNext(Integer t) {
868868
public void merge1AsyncStreamOf1() {
869869
TestSubscriber<Integer> ts = new TestSubscriber<>();
870870
mergeNAsyncStreamsOfN(1, 1).subscribe(ts);
871-
ts.awaitDone(5, TimeUnit.SECONDS);
871+
ts.awaitDone(10, TimeUnit.SECONDS);
872872
ts.assertNoErrors();
873873
assertEquals(1, ts.values().size());
874874
}
@@ -877,7 +877,7 @@ public void merge1AsyncStreamOf1() {
877877
public void merge1AsyncStreamOf1000() {
878878
TestSubscriber<Integer> ts = new TestSubscriber<>();
879879
mergeNAsyncStreamsOfN(1, 1000).subscribe(ts);
880-
ts.awaitDone(5, TimeUnit.SECONDS);
880+
ts.awaitDone(10, TimeUnit.SECONDS);
881881
ts.assertNoErrors();
882882
assertEquals(1000, ts.values().size());
883883
}
@@ -886,7 +886,7 @@ public void merge1AsyncStreamOf1000() {
886886
public void merge10AsyncStreamOf1000() {
887887
TestSubscriber<Integer> ts = new TestSubscriber<>();
888888
mergeNAsyncStreamsOfN(10, 1000).subscribe(ts);
889-
ts.awaitDone(5, TimeUnit.SECONDS);
889+
ts.awaitDone(10, TimeUnit.SECONDS);
890890
ts.assertNoErrors();
891891
assertEquals(10000, ts.values().size());
892892
}
@@ -895,7 +895,7 @@ public void merge10AsyncStreamOf1000() {
895895
public void merge1000AsyncStreamOf1000() {
896896
TestSubscriber<Integer> ts = new TestSubscriber<>();
897897
mergeNAsyncStreamsOfN(1000, 1000).subscribe(ts);
898-
ts.awaitDone(5, TimeUnit.SECONDS);
898+
ts.awaitDone(10, TimeUnit.SECONDS);
899899
ts.assertNoErrors();
900900
assertEquals(1000000, ts.values().size());
901901
}
@@ -904,7 +904,7 @@ public void merge1000AsyncStreamOf1000() {
904904
public void merge2000AsyncStreamOf100() {
905905
TestSubscriber<Integer> ts = new TestSubscriber<>();
906906
mergeNAsyncStreamsOfN(2000, 100).subscribe(ts);
907-
ts.awaitDone(5, TimeUnit.SECONDS);
907+
ts.awaitDone(10, TimeUnit.SECONDS);
908908
ts.assertNoErrors();
909909
assertEquals(200000, ts.values().size());
910910
}
@@ -913,7 +913,7 @@ public void merge2000AsyncStreamOf100() {
913913
public void merge100AsyncStreamOf1() {
914914
TestSubscriber<Integer> ts = new TestSubscriber<>();
915915
mergeNAsyncStreamsOfN(100, 1).subscribe(ts);
916-
ts.awaitDone(5, TimeUnit.SECONDS);
916+
ts.awaitDone(10, TimeUnit.SECONDS);
917917
ts.assertNoErrors();
918918
assertEquals(100, ts.values().size());
919919
}
@@ -935,7 +935,7 @@ public Flowable<Integer> apply(Integer i) {
935935
public void merge1SyncStreamOf1() {
936936
TestSubscriber<Integer> ts = new TestSubscriber<>();
937937
mergeNSyncStreamsOfN(1, 1).subscribe(ts);
938-
ts.awaitDone(5, TimeUnit.SECONDS);
938+
ts.awaitDone(10, TimeUnit.SECONDS);
939939
ts.assertNoErrors();
940940
assertEquals(1, ts.values().size());
941941
}
@@ -944,7 +944,7 @@ public void merge1SyncStreamOf1() {
944944
public void merge1SyncStreamOf1000000() {
945945
TestSubscriber<Integer> ts = new TestSubscriber<>();
946946
mergeNSyncStreamsOfN(1, 1000000).subscribe(ts);
947-
ts.awaitDone(5, TimeUnit.SECONDS);
947+
ts.awaitDone(10, TimeUnit.SECONDS);
948948
ts.assertNoErrors();
949949
assertEquals(1000000, ts.values().size());
950950
}
@@ -953,7 +953,7 @@ public void merge1SyncStreamOf1000000() {
953953
public void merge1000SyncStreamOf1000() {
954954
TestSubscriber<Integer> ts = new TestSubscriber<>();
955955
mergeNSyncStreamsOfN(1000, 1000).subscribe(ts);
956-
ts.awaitDone(5, TimeUnit.SECONDS);
956+
ts.awaitDone(10, TimeUnit.SECONDS);
957957
ts.assertNoErrors();
958958
assertEquals(1000000, ts.values().size());
959959
}
@@ -962,7 +962,7 @@ public void merge1000SyncStreamOf1000() {
962962
public void merge10000SyncStreamOf10() {
963963
TestSubscriber<Integer> ts = new TestSubscriber<>();
964964
mergeNSyncStreamsOfN(10000, 10).subscribe(ts);
965-
ts.awaitDone(5, TimeUnit.SECONDS);
965+
ts.awaitDone(10, TimeUnit.SECONDS);
966966
ts.assertNoErrors();
967967
assertEquals(100000, ts.values().size());
968968
}
@@ -971,7 +971,7 @@ public void merge10000SyncStreamOf10() {
971971
public void merge1000000SyncStreamOf1() {
972972
TestSubscriber<Integer> ts = new TestSubscriber<>();
973973
mergeNSyncStreamsOfN(1000000, 1).subscribe(ts);
974-
ts.awaitDone(5, TimeUnit.SECONDS);
974+
ts.awaitDone(10, TimeUnit.SECONDS);
975975
ts.assertNoErrors();
976976
assertEquals(1000000, ts.values().size());
977977
}
@@ -1043,7 +1043,7 @@ public void subscribe(Subscriber<? super Integer> s) {
10431043

10441044
});
10451045
Flowable.merge(os).subscribe(ts);
1046-
ts.awaitDone(5, TimeUnit.SECONDS);
1046+
ts.awaitDone(10, TimeUnit.SECONDS);
10471047
ts.assertNoErrors();
10481048
assertEquals(10000, ts.values().size());
10491049
}
@@ -1196,7 +1196,7 @@ public void run() {
11961196
latch.countDown();
11971197
}
11981198
}).subscribe();
1199-
boolean a = latch.await(2, TimeUnit.SECONDS);
1199+
boolean a = latch.await(10, TimeUnit.SECONDS);
12001200
if (!a) {
12011201
for (String s : messages) {
12021202
System.out.println("DEBUG => " + s);

‎src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableSubscribeOnTest.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -202,7 +202,7 @@ public void onNext(Integer t) {
202202
System.out.println("First schedule: " + t);
203203
assertTrue(t.getName().startsWith("Rx"));
204204
ts.request(10);
205-
ts.awaitDone(5, TimeUnit.SECONDS);
205+
ts.awaitDone(20, TimeUnit.SECONDS);
206206
System.out.println("After reschedule: " + ts.lastThread());
207207
assertEquals(t, ts.lastThread());
208208
}
@@ -254,7 +254,7 @@ public void onNext(Integer t) {
254254
}
255255

256256
}).subscribeOn(Schedulers.newThread()).subscribe(ts);
257-
ts.awaitDone(5, TimeUnit.SECONDS);
257+
ts.awaitDone(20, TimeUnit.SECONDS);
258258
ts.assertNoErrors();
259259
}
260260

@@ -330,7 +330,7 @@ public void subscribe(FlowableEmitter<Object> s) throws Exception {
330330
.subscribeOn(Schedulers.single())
331331
.observeOn(Schedulers.computation())
332332
.test()
333-
.awaitDone(5, TimeUnit.SECONDS)
333+
.awaitDone(20, TimeUnit.SECONDS)
334334
.assertNoErrors()
335335
.assertComplete();
336336

@@ -355,7 +355,7 @@ public void subscribe(FlowableEmitter<Object> s) throws Exception {
355355
.subscribeOn(Schedulers.single())
356356
.observeOn(Schedulers.computation())
357357
.test()
358-
.awaitDone(5, TimeUnit.SECONDS)
358+
.awaitDone(20, TimeUnit.SECONDS)
359359
.assertValueCount(Flowable.bufferSize())
360360
.assertNoErrors()
361361
.assertComplete();
@@ -377,7 +377,7 @@ public void subscribe(FlowableEmitter<Object> s) throws Exception {
377377
.subscribeOn(Schedulers.single(), false)
378378
.observeOn(Schedulers.computation())
379379
.test()
380-
.awaitDone(5, TimeUnit.SECONDS)
380+
.awaitDone(20, TimeUnit.SECONDS)
381381
.assertNoErrors()
382382
.assertComplete();
383383

@@ -402,7 +402,7 @@ public void subscribe(FlowableEmitter<Object> s) throws Exception {
402402
.subscribeOn(Schedulers.single(), true)
403403
.observeOn(Schedulers.computation())
404404
.test()
405-
.awaitDone(5, TimeUnit.SECONDS)
405+
.awaitDone(20, TimeUnit.SECONDS)
406406
.assertValueCount(Flowable.bufferSize())
407407
.assertNoErrors()
408408
.assertComplete();

‎src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableConcatMapSchedulerTest.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -307,7 +307,7 @@ public Observable<String> apply(Integer t) throws Throwable {
307307
}
308308

309309
@Test
310-
public void issue2890NoStackoverflow() throws InterruptedException {
310+
public void issue2890NoStackoverflow() throws InterruptedException, TimeoutException {
311311
final ExecutorService executor = Executors.newFixedThreadPool(2);
312312
final Scheduler sch = Schedulers.from(executor);
313313

@@ -352,7 +352,11 @@ public void onError(Throwable e) {
352352
}
353353
});
354354

355-
executor.awaitTermination(20000, TimeUnit.MILLISECONDS);
355+
long awaitTerminationTimeout = 100_000;
356+
if (!executor.awaitTermination(awaitTerminationTimeout, TimeUnit.MILLISECONDS)) {
357+
throw new TimeoutException("Completed " + counter.get() + "/" + n + " before timed out after "
358+
+ awaitTerminationTimeout + " milliseconds.");
359+
}
356360

357361
assertEquals(n, counter.get());
358362
}

‎src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableConcatTest.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -683,7 +683,7 @@ public void subscribe(Observer<? super String> observer) {
683683
}
684684

685685
@Test
686-
public void issue2890NoStackoverflow() throws InterruptedException {
686+
public void issue2890NoStackoverflow() throws InterruptedException, TimeoutException {
687687
final ExecutorService executor = Executors.newFixedThreadPool(2);
688688
final Scheduler sch = Schedulers.from(executor);
689689

@@ -728,7 +728,11 @@ public void onError(Throwable e) {
728728
}
729729
});
730730

731-
executor.awaitTermination(20000, TimeUnit.MILLISECONDS);
731+
long awaitTerminationTimeout = 100_000;
732+
if (!executor.awaitTermination(awaitTerminationTimeout, TimeUnit.MILLISECONDS)) {
733+
throw new TimeoutException("Completed " + counter.get() + "/" + n + " before timed out after "
734+
+ awaitTerminationTimeout + " milliseconds.");
735+
}
732736

733737
assertEquals(n, counter.get());
734738
}

‎src/test/java/io/reactivex/rxjava3/subjects/UnicastSubjectTest.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -481,9 +481,7 @@ public void fusedNoConcurrentCleanDueToCancel() {
481481
us.onNext(i);
482482
}
483483

484-
to
485-
.awaitDone(5, TimeUnit.SECONDS)
486-
;
484+
to.awaitDone(10, TimeUnit.SECONDS);
487485

488486
if (!errors.isEmpty()) {
489487
throw new CompositeException(errors);

0 commit comments

Comments
 (0)