diff --git a/src/main/java/io/reactivex/rxjava3/subjects/ReplaySubject.java b/src/main/java/io/reactivex/rxjava3/subjects/ReplaySubject.java
index 4d5eb3335f..e103594ba5 100644
--- a/src/main/java/io/reactivex/rxjava3/subjects/ReplaySubject.java
+++ b/src/main/java/io/reactivex/rxjava3/subjects/ReplaySubject.java
@@ -652,8 +652,6 @@ static final class UnboundedReplayBuffer<T>
 
         final List<Object> buffer;
 
-        volatile boolean done;
-
         volatile int size;
 
         UnboundedReplayBuffer(int capacityHint) {
@@ -671,7 +669,6 @@ public void addFinal(Object notificationLite) {
             buffer.add(notificationLite);
             trimHead();
             size++;
-            done = true;
         }
 
         @Override
@@ -772,20 +769,17 @@ public void replay(ReplayDisposable<T> rs) {
 
                     Object o = b.get(index);
 
-                    if (done) {
-                        if (index + 1 == s) {
-                            s = size;
-                            if (index + 1 == s) {
-                                if (NotificationLite.isComplete(o)) {
-                                    a.onComplete();
-                                } else {
-                                    a.onError(NotificationLite.getError(o));
-                                }
-                                rs.index = null;
-                                rs.cancelled = true;
-                                return;
-                            }
-                        }
+                    if (NotificationLite.isComplete(o)) {
+                        a.onComplete();
+                        rs.index = null;
+                        rs.cancelled = true;
+                        return;
+                    } else
+                    if (NotificationLite.isError(o)) {
+                        a.onError(NotificationLite.getError(o));
+                        rs.index = null;
+                        rs.cancelled = true;
+                        return;
                     }
 
                     a.onNext((T)o);
@@ -856,8 +850,6 @@ static final class SizeBoundReplayBuffer<T>
 
         Node<Object> tail;
 
-        volatile boolean done;
-
         SizeBoundReplayBuffer(int maxSize) {
             this.maxSize = maxSize;
             Node<Object> h = new Node<>(null);
@@ -895,7 +887,6 @@ public void addFinal(Object notificationLite) {
             t.lazySet(n); // releases both the tail and size
 
             trimHead();
-            done = true;
         }
 
         /**
@@ -1000,18 +991,17 @@ public void replay(ReplayDisposable<T> rs) {
 
                     Object o = n.value;
 
-                    if (done) {
-                        if (n.get() == null) {
-
-                            if (NotificationLite.isComplete(o)) {
-                                a.onComplete();
-                            } else {
-                                a.onError(NotificationLite.getError(o));
-                            }
-                            rs.index = null;
-                            rs.cancelled = true;
-                            return;
-                        }
+                    if (NotificationLite.isComplete(o)) {
+                        a.onComplete();
+                        rs.index = null;
+                        rs.cancelled = true;
+                        return;
+                    } else
+                    if (NotificationLite.isError(o)) {
+                        a.onError(NotificationLite.getError(o));
+                        rs.index = null;
+                        rs.cancelled = true;
+                        return;
                     }
 
                     a.onNext((T)o);
@@ -1069,8 +1059,6 @@ static final class SizeAndTimeBoundReplayBuffer<T>
 
         TimedNode<Object> tail;
 
-        volatile boolean done;
-
         SizeAndTimeBoundReplayBuffer(int maxSize, long maxAge, TimeUnit unit, Scheduler scheduler) {
             this.maxSize = maxSize;
             this.maxAge = maxAge;
@@ -1163,8 +1151,6 @@ public void addFinal(Object notificationLite) {
             size++;
             t.lazySet(n); // releases both the tail and size
             trimFinal();
-
-            done = true;
         }
 
         /**
@@ -1290,18 +1276,17 @@ public void replay(ReplayDisposable<T> rs) {
 
                     Object o = n.value;
 
-                    if (done) {
-                        if (n.get() == null) {
-
-                            if (NotificationLite.isComplete(o)) {
-                                a.onComplete();
-                            } else {
-                                a.onError(NotificationLite.getError(o));
-                            }
-                            rs.index = null;
-                            rs.cancelled = true;
-                            return;
-                        }
+                    if (NotificationLite.isComplete(o)) {
+                        a.onComplete();
+                        rs.index = null;
+                        rs.cancelled = true;
+                        return;
+                    } else
+                    if (NotificationLite.isError(o)) {
+                        a.onError(NotificationLite.getError(o));
+                        rs.index = null;
+                        rs.cancelled = true;
+                        return;
                     }
 
                     a.onNext((T)o);
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableAmbTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableAmbTest.java
index ba86c69719..1a1ceca926 100644
--- a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableAmbTest.java
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableAmbTest.java
@@ -402,7 +402,8 @@ public void disposed() {
 
     @Test
     public void manySources() {
-        Flowable<?>[] a = new Flowable[32];
+        @SuppressWarnings("unchecked")
+        Flowable<Object>[] a = new Flowable[32];
         Arrays.fill(a, Flowable.never());
         a[31] = Flowable.just(1);
 
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableAmbTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableAmbTest.java
index 0a349cd417..5ec9129d9f 100644
--- a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableAmbTest.java
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableAmbTest.java
@@ -235,7 +235,8 @@ public void ambArraySingleElement() {
 
     @Test
     public void manySources() {
-        Observable<?>[] a = new Observable[32];
+        @SuppressWarnings("unchecked")
+        Observable<Object>[] a = new Observable[32];
         Arrays.fill(a, Observable.never());
         a[31] = Observable.just(1);
 
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleAmbTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleAmbTest.java
index 083ee46238..8086aa7c1c 100644
--- a/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleAmbTest.java
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/single/SingleAmbTest.java
@@ -249,7 +249,8 @@ public void run() {
 
     @Test
     public void manySources() {
-        Single<?>[] sources = new Single[32];
+        @SuppressWarnings("unchecked")
+        Single<Object>[] sources = new Single[32];
         Arrays.fill(sources, Single.never());
         sources[31] = Single.just(31);
 
diff --git a/src/test/java/io/reactivex/rxjava3/processors/ReplayProcessorTest.java b/src/test/java/io/reactivex/rxjava3/processors/ReplayProcessorTest.java
index aa7026a9ff..35a877911b 100644
--- a/src/test/java/io/reactivex/rxjava3/processors/ReplayProcessorTest.java
+++ b/src/test/java/io/reactivex/rxjava3/processors/ReplayProcessorTest.java
@@ -1832,4 +1832,69 @@ public void timeAndSizeRemoveCorrectNumberOfOld() {
 
         rp.test().assertValuesOnly(4, 5);
     }
-}
+
+    @Test
+    public void terminationSubscriptionRaceUnbounded() throws Throwable {
+        for (int i = 1; i <= 10000; i++) {
+            ReplayProcessor<String> source = ReplayProcessor.create();
+            PublishProcessor<String> sink = PublishProcessor.create();
+            TestSubscriber<String> subscriber = sink.test();
+            Schedulers.computation().scheduleDirect(() -> {
+                // issue signals to the source in adherence to the reactive streams specification
+                source.onSubscribe(new BooleanSubscription());
+                source.onNext("hello");
+                source.onNext("world");
+                source.onComplete();
+            });
+            Schedulers.computation().scheduleDirect(() -> {
+                // connect the source to the sink in parallel with the signals issued to the source
+                // note the cast() operator, which is here to detect non-String escapees
+                source.cast(String.class).subscribe(sink);
+            });
+            subscriber.await().assertValues("hello", "world").assertComplete();
+        }
+    }
+
+    @Test
+    public void terminationSubscriptionRaceSizeBound() throws Throwable {
+        for (int i = 1; i <= 10000; i++) {
+            ReplayProcessor<String> source = ReplayProcessor.createWithSize(20);
+            PublishProcessor<String> sink = PublishProcessor.create();
+            TestSubscriber<String> subscriber = sink.test();
+            Schedulers.computation().scheduleDirect(() -> {
+                // issue signals to the source in adherence to the reactive streams specification
+                source.onSubscribe(new BooleanSubscription());
+                source.onNext("hello");
+                source.onNext("world");
+                source.onComplete();
+            });
+            Schedulers.computation().scheduleDirect(() -> {
+                // connect the source to the sink in parallel with the signals issued to the source
+                // note the cast() operator, which is here to detect non-String escapees
+                source.cast(String.class).subscribe(sink);
+            });
+            subscriber.await().assertValues("hello", "world").assertComplete();
+        }
+    }
+
+    @Test
+    public void terminationSubscriptionRaceTimeBound() throws Throwable {
+        for (int i = 1; i <= 10000; i++) {
+            ReplayProcessor<String> source = ReplayProcessor.createWithTime(20, TimeUnit.MINUTES, Schedulers.computation());
+            PublishProcessor<String> sink = PublishProcessor.create();
+            TestSubscriber<String> subscriber = sink.test();
+            Schedulers.computation().scheduleDirect(() -> {
+                // issue signals to the source in adherence to the reactive streams specification
+                source.onSubscribe(new BooleanSubscription());
+                source.onNext("hello");
+                source.onNext("world");
+                source.onComplete();
+            });
+            Schedulers.computation().scheduleDirect(() -> {
+                // connect the source to the sink in parallel with the signals issued to the source
+                // note the cast() operator, which is here to detect non-String escapees
+                source.cast(String.class).subscribe(sink);
+            });
+            subscriber.await().assertValues("hello", "world").assertComplete();
+        }
+    }}
diff --git a/src/test/java/io/reactivex/rxjava3/subjects/ReplaySubjectTest.java b/src/test/java/io/reactivex/rxjava3/subjects/ReplaySubjectTest.java
index e752278b2f..8417b53081 100644
--- a/src/test/java/io/reactivex/rxjava3/subjects/ReplaySubjectTest.java
+++ b/src/test/java/io/reactivex/rxjava3/subjects/ReplaySubjectTest.java
@@ -1378,4 +1378,70 @@ public void timeAndSizeRemoveCorrectNumberOfOld() {
 
         rs.test().assertValuesOnly(4, 5);
     }
+
+    @Test
+    public void terminationSubscriptionRaceUnbounded() throws Throwable {
+        for (int i = 1; i <= 10000; i++) {
+            Subject<String> source = ReplaySubject.create();
+            Subject<String> sink = PublishSubject.create();
+            TestObserver<String> observer = sink.test();
+            Schedulers.computation().scheduleDirect(() -> {
+                // issue signals to the source in adherence to the reactive streams specification
+                source.onSubscribe(Disposable.empty());
+                source.onNext("hello");
+                source.onNext("world");
+                source.onComplete();
+            });
+            Schedulers.computation().scheduleDirect(() -> {
+                // connect the source to the sink in parallel with the signals issued to the source
+                // note the cast() operator, which is here to detect non-String escapees
+                source.cast(String.class).subscribe(sink);
+            });
+            observer.await().assertValues("hello", "world").assertComplete();
+        }
+    }
+
+    @Test
+    public void terminationSubscriptionRaceSizeBound() throws Throwable {
+        for (int i = 1; i <= 10000; i++) {
+            Subject<String> source = ReplaySubject.createWithSize(20);
+            Subject<String> sink = PublishSubject.create();
+            TestObserver<String> observer = sink.test();
+            Schedulers.computation().scheduleDirect(() -> {
+                // issue signals to the source in adherence to the reactive streams specification
+                source.onSubscribe(Disposable.empty());
+                source.onNext("hello");
+                source.onNext("world");
+                source.onComplete();
+            });
+            Schedulers.computation().scheduleDirect(() -> {
+                // connect the source to the sink in parallel with the signals issued to the source
+                // note the cast() operator, which is here to detect non-String escapees
+                source.cast(String.class).subscribe(sink);
+            });
+            observer.await().assertValues("hello", "world").assertComplete();
+        }
+    }
+
+    @Test
+    public void terminationSubscriptionRaceTimeBound() throws Throwable {
+        for (int i = 1; i <= 10000; i++) {
+            Subject<String> source = ReplaySubject.createWithTime(20, TimeUnit.MINUTES, Schedulers.computation());
+            Subject<String> sink = PublishSubject.create();
+            TestObserver<String> observer = sink.test();
+            Schedulers.computation().scheduleDirect(() -> {
+                // issue signals to the source in adherence to the reactive streams specification
+                source.onSubscribe(Disposable.empty());
+                source.onNext("hello");
+                source.onNext("world");
+                source.onComplete();
+            });
+            Schedulers.computation().scheduleDirect(() -> {
+                // connect the source to the sink in parallel with the signals issued to the source
+                // note the cast() operator, which is here to detect non-String escapees
+                source.cast(String.class).subscribe(sink);
+            });
+            observer.await().assertValues("hello", "world").assertComplete();
+        }
+    }
 }