diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatArray.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatArray.java
index c06cffa80a..4d7cd06c5b 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatArray.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatArray.java
@@ -59,6 +59,7 @@ static final class ConcatArraySubscriber<T> extends SubscriptionArbiter implemen
         long produced;
 
         ConcatArraySubscriber(Publisher<? extends T>[] sources, boolean delayError, Subscriber<? super T> downstream) {
+            super(false);
             this.downstream = downstream;
             this.sources = sources;
             this.delayError = delayError;
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java
index edde82442e..2dc316d1e9 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java
@@ -571,6 +571,7 @@ static final class ConcatMapInner<R>
         long produced;
 
         ConcatMapInner(ConcatMapSupport<R> parent) {
+            super(false);
             this.parent = parent;
         }
 
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableDelaySubscriptionOther.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableDelaySubscriptionOther.java
index a98892a01c..ea52987d59 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableDelaySubscriptionOther.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableDelaySubscriptionOther.java
@@ -12,10 +12,12 @@
  */
 package io.reactivex.internal.operators.flowable;
 
+import java.util.concurrent.atomic.*;
+
 import org.reactivestreams.*;
 
 import io.reactivex.*;
-import io.reactivex.internal.subscriptions.SubscriptionArbiter;
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
@@ -35,94 +37,105 @@ public FlowableDelaySubscriptionOther(Publisher<? extends T> main, Publisher<U>
 
     @Override
     public void subscribeActual(final Subscriber<? super T> child) {
-        final SubscriptionArbiter serial = new SubscriptionArbiter();
-        child.onSubscribe(serial);
+        MainSubscriber<T> parent = new MainSubscriber<T>(child, main);
+        child.onSubscribe(parent);
+        other.subscribe(parent.other);
+    }
 
-        FlowableSubscriber<U> otherSubscriber = new DelaySubscriber(serial, child);
+    static final class MainSubscriber<T> extends AtomicLong implements FlowableSubscriber<T>, Subscription {
 
-        other.subscribe(otherSubscriber);
-    }
+        private static final long serialVersionUID = 2259811067697317255L;
+
+        final Subscriber<? super T> downstream;
 
-    final class DelaySubscriber implements FlowableSubscriber<U> {
-        final SubscriptionArbiter serial;
-        final Subscriber<? super T> child;
-        boolean done;
+        final Publisher<? extends T> main;
 
-        DelaySubscriber(SubscriptionArbiter serial, Subscriber<? super T> child) {
-            this.serial = serial;
-            this.child = child;
+        final OtherSubscriber other;
+
+        final AtomicReference<Subscription> upstream;
+
+        MainSubscriber(Subscriber<? super T> downstream, Publisher<? extends T> main) {
+            this.downstream = downstream;
+            this.main = main;
+            this.other = new OtherSubscriber();
+            this.upstream = new AtomicReference<Subscription>();
         }
 
-        @Override
-        public void onSubscribe(final Subscription s) {
-            serial.setSubscription(new DelaySubscription(s));
-            s.request(Long.MAX_VALUE);
+        void next() {
+            main.subscribe(this);
         }
 
         @Override
-        public void onNext(U t) {
-            onComplete();
+        public void onNext(T t) {
+            downstream.onNext(t);
         }
 
         @Override
-        public void onError(Throwable e) {
-            if (done) {
-                RxJavaPlugins.onError(e);
-                return;
-            }
-            done = true;
-            child.onError(e);
+        public void onError(Throwable t) {
+            downstream.onError(t);
         }
 
         @Override
         public void onComplete() {
-            if (done) {
-                return;
-            }
-            done = true;
-
-            main.subscribe(new OnCompleteSubscriber());
+            downstream.onComplete();
         }
 
-        final class DelaySubscription implements Subscription {
-
-            final Subscription upstream;
-
-            DelaySubscription(Subscription s) {
-                this.upstream = s;
+        @Override
+        public void request(long n) {
+            if (SubscriptionHelper.validate(n)) {
+                SubscriptionHelper.deferredRequest(upstream, this, n);
             }
+        }
 
-            @Override
-            public void request(long n) {
-                // ignored
-            }
+        @Override
+        public void cancel() {
+            SubscriptionHelper.cancel(other);
+            SubscriptionHelper.cancel(upstream);
+        }
 
-            @Override
-            public void cancel() {
-                upstream.cancel();
-            }
+        @Override
+        public void onSubscribe(Subscription s) {
+            SubscriptionHelper.deferredSetOnce(upstream, this, s);
         }
 
-        final class OnCompleteSubscriber implements FlowableSubscriber<T> {
+        final class OtherSubscriber extends AtomicReference<Subscription> implements FlowableSubscriber<Object> {
+
+            private static final long serialVersionUID = -3892798459447644106L;
+
             @Override
             public void onSubscribe(Subscription s) {
-                serial.setSubscription(s);
+                if (SubscriptionHelper.setOnce(this, s)) {
+                    s.request(Long.MAX_VALUE);
+                }
             }
 
             @Override
-            public void onNext(T t) {
-                child.onNext(t);
+            public void onNext(Object t) {
+                Subscription s = get();
+                if (s != SubscriptionHelper.CANCELLED) {
+                    lazySet(SubscriptionHelper.CANCELLED);
+                    s.cancel();
+                    next();
+                }
             }
 
             @Override
             public void onError(Throwable t) {
-                child.onError(t);
+                Subscription s = get();
+                if (s != SubscriptionHelper.CANCELLED) {
+                    downstream.onError(t);
+                } else {
+                    RxJavaPlugins.onError(t);
+                }
             }
 
             @Override
             public void onComplete() {
-                child.onComplete();
+                Subscription s = get();
+                if (s != SubscriptionHelper.CANCELLED) {
+                    next();
+                }
             }
         }
-    }
+}
 }
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnErrorNext.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnErrorNext.java
index 4d5b86c880..7110d086e0 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnErrorNext.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnErrorNext.java
@@ -58,6 +58,7 @@ static final class OnErrorNextSubscriber<T>
         long produced;
 
         OnErrorNextSubscriber(Subscriber<? super T> actual, Function<? super Throwable, ? extends Publisher<? extends T>> nextSupplier, boolean allowFatal) {
+            super(false);
             this.downstream = actual;
             this.nextSupplier = nextSupplier;
             this.allowFatal = allowFatal;
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeat.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeat.java
index f99bc48a20..1bbdd7cd5c 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeat.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeat.java
@@ -29,7 +29,7 @@ public FlowableRepeat(Flowable<T> source, long count) {
 
     @Override
     public void subscribeActual(Subscriber<? super T> s) {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(false);
         s.onSubscribe(sa);
 
         RepeatSubscriber<T> rs = new RepeatSubscriber<T>(s, count != Long.MAX_VALUE ? count - 1 : Long.MAX_VALUE, sa, source);
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeatUntil.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeatUntil.java
index d373a3865d..9c7057af59 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeatUntil.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeatUntil.java
@@ -31,7 +31,7 @@ public FlowableRepeatUntil(Flowable<T> source, BooleanSupplier until) {
 
     @Override
     public void subscribeActual(Subscriber<? super T> s) {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(false);
         s.onSubscribe(sa);
 
         RepeatSubscriber<T> rs = new RepeatSubscriber<T>(s, until, sa, source);
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeatWhen.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeatWhen.java
index c5fac022ad..45f8bfbb1e 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeatWhen.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRepeatWhen.java
@@ -143,6 +143,7 @@ abstract static class WhenSourceSubscriber<T, U> extends SubscriptionArbiter imp
 
         WhenSourceSubscriber(Subscriber<? super T> actual, FlowableProcessor<U> processor,
                 Subscription receiver) {
+            super(false);
             this.downstream = actual;
             this.processor = processor;
             this.receiver = receiver;
@@ -160,6 +161,7 @@ public final void onNext(T t) {
         }
 
         protected final void again(U signal) {
+            setSubscription(EmptySubscription.INSTANCE);
             long p = produced;
             if (p != 0L) {
                 produced = 0L;
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRetryBiPredicate.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRetryBiPredicate.java
index 662ddb694c..8bc9ba27d0 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRetryBiPredicate.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRetryBiPredicate.java
@@ -33,7 +33,7 @@ public FlowableRetryBiPredicate(
 
     @Override
     public void subscribeActual(Subscriber<? super T> s) {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(false);
         s.onSubscribe(sa);
 
         RetryBiSubscriber<T> rs = new RetryBiSubscriber<T>(s, predicate, sa, source);
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRetryPredicate.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRetryPredicate.java
index 47a8e3d878..8d035aaf95 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRetryPredicate.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRetryPredicate.java
@@ -35,7 +35,7 @@ public FlowableRetryPredicate(Flowable<T> source,
 
     @Override
     public void subscribeActual(Subscriber<? super T> s) {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(false);
         s.onSubscribe(sa);
 
         RetrySubscriber<T> rs = new RetrySubscriber<T>(s, count, predicate, sa, source);
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchIfEmpty.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchIfEmpty.java
index f9fc7ebc4c..be8febdf17 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchIfEmpty.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchIfEmpty.java
@@ -43,7 +43,7 @@ static final class SwitchIfEmptySubscriber<T> implements FlowableSubscriber<T> {
             this.downstream = actual;
             this.other = other;
             this.empty = true;
-            this.arbiter = new SubscriptionArbiter();
+            this.arbiter = new SubscriptionArbiter(false);
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.java
index 6d8300f234..eec781bcc3 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.java
@@ -208,6 +208,7 @@ static final class TimeoutFallbackSubscriber<T> extends SubscriptionArbiter
         TimeoutFallbackSubscriber(Subscriber<? super T> actual,
                 Function<? super T, ? extends Publisher<?>> itemTimeoutIndicator,
                         Publisher<? extends T> fallback) {
+            super(true);
             this.downstream = actual;
             this.itemTimeoutIndicator = itemTimeoutIndicator;
             this.task = new SequentialDisposable();
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java
index 54a2762dfc..d25acdc3e3 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java
@@ -196,6 +196,7 @@ static final class TimeoutFallbackSubscriber<T> extends SubscriptionArbiter
 
         TimeoutFallbackSubscriber(Subscriber<? super T> actual, long timeout, TimeUnit unit,
                 Scheduler.Worker worker, Publisher<? extends T> fallback) {
+            super(true);
             this.downstream = actual;
             this.timeout = timeout;
             this.unit = unit;
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMap.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMap.java
index 831cfd1c55..a59841d501 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMap.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableConcatMap.java
@@ -244,7 +244,7 @@ static final class InnerObserver<U> extends AtomicReference<Disposable> implemen
 
             @Override
             public void onSubscribe(Disposable d) {
-                DisposableHelper.set(this, d);
+                DisposableHelper.replace(this, d);
             }
 
             @Override
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableRepeatWhen.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableRepeatWhen.java
index 4b8e194973..af27f2f6d0 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableRepeatWhen.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableRepeatWhen.java
@@ -92,7 +92,7 @@ static final class RepeatWhenObserver<T> extends AtomicInteger implements Observ
 
         @Override
         public void onSubscribe(Disposable d) {
-            DisposableHelper.replace(this.upstream, d);
+            DisposableHelper.setOnce(this.upstream, d);
         }
 
         @Override
@@ -109,6 +109,7 @@ public void onError(Throwable e) {
         @Override
         public void onComplete() {
             active = false;
+            DisposableHelper.replace(upstream, null);
             signaller.onNext(0);
         }
 
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryBiPredicate.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryBiPredicate.java
index c9be2c75ca..f762142f0b 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryBiPredicate.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryBiPredicate.java
@@ -58,7 +58,7 @@ static final class RetryBiObserver<T> extends AtomicInteger implements Observer<
 
         @Override
         public void onSubscribe(Disposable d) {
-            upstream.update(d);
+            upstream.replace(d);
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryPredicate.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryPredicate.java
index 6fb5d7b665..ee5e074f58 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryPredicate.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryPredicate.java
@@ -61,7 +61,7 @@ static final class RepeatObserver<T> extends AtomicInteger implements Observer<T
 
         @Override
         public void onSubscribe(Disposable d) {
-            upstream.update(d);
+            upstream.replace(d);
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryWhen.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryWhen.java
index 65a17fe269..0d48ef1239 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryWhen.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableRetryWhen.java
@@ -102,6 +102,7 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable e) {
+            DisposableHelper.replace(upstream, null);
             active = false;
             signaller.onNext(e);
         }
diff --git a/src/main/java/io/reactivex/internal/subscriptions/SubscriptionArbiter.java b/src/main/java/io/reactivex/internal/subscriptions/SubscriptionArbiter.java
index 6abda2ce51..2796573392 100644
--- a/src/main/java/io/reactivex/internal/subscriptions/SubscriptionArbiter.java
+++ b/src/main/java/io/reactivex/internal/subscriptions/SubscriptionArbiter.java
@@ -55,11 +55,14 @@ public class SubscriptionArbiter extends AtomicInteger implements Subscription {
 
     final AtomicLong missedProduced;
 
+    final boolean cancelOnReplace;
+
     volatile boolean cancelled;
 
     protected boolean unbounded;
 
-    public SubscriptionArbiter() {
+    public SubscriptionArbiter(boolean cancelOnReplace) {
+        this.cancelOnReplace = cancelOnReplace;
         missedSubscription = new AtomicReference<Subscription>();
         missedRequested = new AtomicLong();
         missedProduced = new AtomicLong();
@@ -80,7 +83,7 @@ public final void setSubscription(Subscription s) {
         if (get() == 0 && compareAndSet(0, 1)) {
             Subscription a = actual;
 
-            if (a != null) {
+            if (a != null && cancelOnReplace) {
                 a.cancel();
             }
 
@@ -100,7 +103,7 @@ public final void setSubscription(Subscription s) {
         }
 
         Subscription a = missedSubscription.getAndSet(s);
-        if (a != null) {
+        if (a != null && cancelOnReplace) {
             a.cancel();
         }
         drain();
@@ -240,7 +243,7 @@ final void drainLoop() {
                 }
 
                 if (ms != null) {
-                    if (a != null) {
+                    if (a != null && cancelOnReplace) {
                         a.cancel();
                     }
                     actual = ms;
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapTest.java
index ac5c573910..eba09e564f 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapTest.java
@@ -13,14 +13,17 @@
 
 package io.reactivex.internal.operators.flowable;
 
+import static org.junit.Assert.assertEquals;
+
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 
 import io.reactivex.*;
 import io.reactivex.exceptions.TestException;
-import io.reactivex.functions.Function;
+import io.reactivex.functions.*;
 import io.reactivex.internal.operators.flowable.FlowableConcatMap.WeakScalarSubscription;
 import io.reactivex.schedulers.Schedulers;
 import io.reactivex.subscribers.TestSubscriber;
@@ -143,4 +146,26 @@ public Publisher<Integer> apply(Integer v)
         .test()
         .assertFailure(TestException.class);
     }
+
+    @Test
+    public void noCancelPrevious() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Flowable.range(1, 5)
+        .concatMap(new Function<Integer, Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> apply(Integer v) throws Exception {
+                return Flowable.just(v).doOnCancel(new Action() {
+                    @Override
+                    public void run() throws Exception {
+                        counter.getAndIncrement();
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1, 2, 3, 4, 5);
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatTest.java
index 733f3b7cf3..772f559733 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatTest.java
@@ -29,7 +29,7 @@
 import io.reactivex.*;
 import io.reactivex.disposables.*;
 import io.reactivex.exceptions.*;
-import io.reactivex.functions.Function;
+import io.reactivex.functions.*;
 import io.reactivex.internal.functions.Functions;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -1627,4 +1627,42 @@ public void subscribe(FlowableEmitter<Integer> s) throws Exception {
 
         assertEquals(1, calls[0]);
     }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void noCancelPreviousArray() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.just(1).doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        Flowable.concatArray(source, source, source, source, source)
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void noCancelPreviousIterable() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.just(1).doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        Flowable.concat(Arrays.asList(source, source, source, source, source))
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRepeatTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRepeatTest.java
index 82cec69233..51376bc594 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRepeatTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRepeatTest.java
@@ -368,4 +368,77 @@ public Flowable<Object> apply(Flowable<Object> handler) throws Exception {
         .test()
         .assertResult(1, 2, 3, 1, 2, 3);
     }
+
+    @Test
+    public void noCancelPreviousRepeat() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.just(1).doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.repeat(5)
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRepeatUntil() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.just(1).doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        final AtomicInteger times = new AtomicInteger();
+
+        source.repeatUntil(new BooleanSupplier() {
+            @Override
+            public boolean getAsBoolean() throws Exception {
+                return times.getAndIncrement() == 4;
+            }
+        })
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRepeatWhen() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.just(1).doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        final AtomicInteger times = new AtomicInteger();
+
+        source.repeatWhen(new Function<Flowable<Object>, Flowable<?>>() {
+            @Override
+            public Flowable<?> apply(Flowable<Object> e) throws Exception {
+                return e.takeWhile(new Predicate<Object>() {
+                    @Override
+                    public boolean test(Object v) throws Exception {
+                        return times.getAndIncrement() < 4;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRetryTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRetryTest.java
index 8a0a29f8dd..a5dd4918ab 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRetryTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRetryTest.java
@@ -30,6 +30,7 @@
 import io.reactivex.exceptions.TestException;
 import io.reactivex.flowables.GroupedFlowable;
 import io.reactivex.functions.*;
+import io.reactivex.internal.functions.Functions;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
 import io.reactivex.processors.PublishProcessor;
 import io.reactivex.schedulers.Schedulers;
@@ -355,7 +356,7 @@ public void testRetrySubscribesAgainAfterError() throws Exception {
         Consumer<Integer> throwException = mock(Consumer.class);
         doThrow(new RuntimeException()).when(throwException).accept(Mockito.anyInt());
 
-        // create a retrying observable based on a PublishProcessor
+        // create a retrying Flowable based on a PublishProcessor
         PublishProcessor<Integer> processor = PublishProcessor.create();
         processor
         // record item
@@ -492,7 +493,7 @@ public void cancel() {
     }
 
     @Test
-    public void testSourceObservableCallsUnsubscribe() throws InterruptedException {
+    public void testSourceFlowableCallsUnsubscribe() throws InterruptedException {
         final AtomicInteger subsCount = new AtomicInteger(0);
 
         final TestSubscriber<String> ts = new TestSubscriber<String>();
@@ -523,7 +524,7 @@ public void subscribe(Subscriber<? super String> s) {
     }
 
     @Test
-    public void testSourceObservableRetry1() throws InterruptedException {
+    public void testSourceFlowableRetry1() throws InterruptedException {
         final AtomicInteger subsCount = new AtomicInteger(0);
 
         final TestSubscriber<String> ts = new TestSubscriber<String>();
@@ -542,7 +543,7 @@ public void subscribe(Subscriber<? super String> s) {
     }
 
     @Test
-    public void testSourceObservableRetry0() throws InterruptedException {
+    public void testSourceFlowableRetry0() throws InterruptedException {
         final AtomicInteger subsCount = new AtomicInteger(0);
 
         final TestSubscriber<String> ts = new TestSubscriber<String>();
@@ -566,12 +567,14 @@ static final class SlowFlowable implements Publisher<Long> {
         final AtomicInteger active = new AtomicInteger(0);
         final AtomicInteger maxActive = new AtomicInteger(0);
         final AtomicInteger nextBeforeFailure;
+        final String context;
 
         private final int emitDelay;
 
-        SlowFlowable(int emitDelay, int countNext) {
+        SlowFlowable(int emitDelay, int countNext, String context) {
             this.emitDelay = emitDelay;
             this.nextBeforeFailure = new AtomicInteger(countNext);
+            this.context = context;
         }
 
         @Override
@@ -593,7 +596,7 @@ public void cancel() {
             efforts.getAndIncrement();
             active.getAndIncrement();
             maxActive.set(Math.max(active.get(), maxActive.get()));
-            final Thread thread = new Thread() {
+            final Thread thread = new Thread(context) {
                 @Override
                 public void run() {
                     long nr = 0;
@@ -603,7 +606,9 @@ public void run() {
                             if (nextBeforeFailure.getAndDecrement() > 0) {
                                 subscriber.onNext(nr++);
                             } else {
+                                active.decrementAndGet();
                                 subscriber.onError(new RuntimeException("expected-failed"));
+                                break;
                             }
                         }
                     } catch (InterruptedException t) {
@@ -664,7 +669,7 @@ public void testUnsubscribeAfterError() {
         Subscriber<Long> subscriber = TestHelper.mockSubscriber();
 
         // Flowable that always fails after 100ms
-        SlowFlowable so = new SlowFlowable(100, 0);
+        SlowFlowable so = new SlowFlowable(100, 0, "testUnsubscribeAfterError");
         Flowable<Long> f = Flowable.unsafeCreate(so).retry(5);
 
         AsyncSubscriber<Long> async = new AsyncSubscriber<Long>(subscriber);
@@ -688,7 +693,7 @@ public void testTimeoutWithRetry() {
         Subscriber<Long> subscriber = TestHelper.mockSubscriber();
 
         // Flowable that sends every 100ms (timeout fails instead)
-        SlowFlowable sf = new SlowFlowable(100, 10);
+        SlowFlowable sf = new SlowFlowable(100, 10, "testTimeoutWithRetry");
         Flowable<Long> f = Flowable.unsafeCreate(sf).timeout(80, TimeUnit.MILLISECONDS).retry(5);
 
         AsyncSubscriber<Long> async = new AsyncSubscriber<Long>(subscriber);
@@ -1001,7 +1006,7 @@ public boolean getAsBoolean() throws Exception {
     }
 
     @Test
-    public void shouldDisposeInnerObservable() {
+    public void shouldDisposeInnerFlowable() {
       final PublishProcessor<Object> processor = PublishProcessor.create();
       final Disposable disposable = Flowable.error(new RuntimeException("Leak"))
           .retryWhen(new Function<Flowable<Throwable>, Flowable<Object>>() {
@@ -1021,4 +1026,199 @@ public Flowable<Object> apply(Throwable ignore) throws Exception {
       disposable.dispose();
       assertFalse(processor.hasSubscribers());
     }
+
+    @Test
+    public void noCancelPreviousRetry() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.defer(new Callable<Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> call() throws Exception {
+                if (times.getAndIncrement() < 4) {
+                    return Flowable.error(new TestException());
+                }
+                return Flowable.just(1);
+            }
+        })
+        .doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retry(5)
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRetryWhile() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.defer(new Callable<Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> call() throws Exception {
+                if (times.getAndIncrement() < 4) {
+                    return Flowable.error(new TestException());
+                }
+                return Flowable.just(1);
+            }
+        })
+        .doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retry(5, Functions.alwaysTrue())
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRetryWhile2() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.defer(new Callable<Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> call() throws Exception {
+                if (times.getAndIncrement() < 4) {
+                    return Flowable.error(new TestException());
+                }
+                return Flowable.just(1);
+            }
+        })
+        .doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retry(new BiPredicate<Integer, Throwable>() {
+            @Override
+            public boolean test(Integer a, Throwable b) throws Exception {
+                return a < 5;
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRetryUntil() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.defer(new Callable<Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> call() throws Exception {
+                if (times.getAndIncrement() < 4) {
+                    return Flowable.error(new TestException());
+                }
+                return Flowable.just(1);
+            }
+        })
+        .doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retryUntil(new BooleanSupplier() {
+            @Override
+            public boolean getAsBoolean() throws Exception {
+                return false;
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRepeatWhen() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.defer(new Callable<Flowable<Integer>>() {
+            @Override
+            public Flowable<Integer> call() throws Exception {
+                if (times.get() < 4) {
+                    return Flowable.error(new TestException());
+                }
+                return Flowable.just(1);
+            }
+        }).doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retryWhen(new Function<Flowable<Throwable>, Flowable<?>>() {
+            @Override
+            public Flowable<?> apply(Flowable<Throwable> e) throws Exception {
+                return e.takeWhile(new Predicate<Object>() {
+                    @Override
+                    public boolean test(Object v) throws Exception {
+                        return times.getAndIncrement() < 4;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRepeatWhen2() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Flowable<Integer> source = Flowable.<Integer>error(new TestException())
+                .doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retryWhen(new Function<Flowable<Throwable>, Flowable<?>>() {
+            @Override
+            public Flowable<?> apply(Flowable<Throwable> e) throws Exception {
+                return e.takeWhile(new Predicate<Object>() {
+                    @Override
+                    public boolean test(Object v) throws Exception {
+                        return times.getAndIncrement() < 4;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult();
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRetryWithPredicateTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRetryWithPredicateTest.java
index 54346a73c0..de31e283a4 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRetryWithPredicateTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRetryWithPredicateTest.java
@@ -230,7 +230,7 @@ public void testUnsubscribeAfterError() {
         Subscriber<Long> subscriber = TestHelper.mockSubscriber();
 
         // Flowable that always fails after 100ms
-        FlowableRetryTest.SlowFlowable so = new FlowableRetryTest.SlowFlowable(100, 0);
+        FlowableRetryTest.SlowFlowable so = new FlowableRetryTest.SlowFlowable(100, 0, "testUnsubscribeAfterError");
         Flowable<Long> f = Flowable
                 .unsafeCreate(so)
                 .retry(retry5);
@@ -256,7 +256,7 @@ public void testTimeoutWithRetry() {
         Subscriber<Long> subscriber = TestHelper.mockSubscriber();
 
         // Flowable that sends every 100ms (timeout fails instead)
-        FlowableRetryTest.SlowFlowable so = new FlowableRetryTest.SlowFlowable(100, 10);
+        FlowableRetryTest.SlowFlowable so = new FlowableRetryTest.SlowFlowable(100, 10, "testTimeoutWithRetry");
         Flowable<Long> f = Flowable
                 .unsafeCreate(so)
                 .timeout(80, TimeUnit.MILLISECONDS)
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchIfEmptyTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchIfEmptyTest.java
index 99ecaaac43..f429d8e172 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchIfEmptyTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchIfEmptyTest.java
@@ -127,7 +127,7 @@ public void onNext(Long aLong) {
     }
 
     @Test
-    public void testSwitchShouldTriggerUnsubscribe() {
+    public void testSwitchShouldNotTriggerUnsubscribe() {
         final BooleanSubscription bs = new BooleanSubscription();
 
         Flowable.unsafeCreate(new Publisher<Long>() {
@@ -137,7 +137,7 @@ public void subscribe(final Subscriber<? super Long> subscriber) {
                 subscriber.onComplete();
             }
         }).switchIfEmpty(Flowable.<Long>never()).subscribe();
-        assertTrue(bs.isCancelled());
+        assertFalse(bs.isCancelled());
     }
 
     @Test
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapTest.java
index 5c2681a79d..4940d6c857 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapTest.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatMapTest.java
@@ -13,17 +13,18 @@
 
 package io.reactivex.internal.operators.observable;
 
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
 import java.util.List;
 import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
 import io.reactivex.*;
 import io.reactivex.disposables.*;
 import io.reactivex.exceptions.*;
-import io.reactivex.functions.Function;
+import io.reactivex.functions.*;
 import io.reactivex.internal.functions.Functions;
 import io.reactivex.observers.TestObserver;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -498,4 +499,26 @@ public void onNext(Integer t) {
 
         to.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
     }
+
+    @Test
+    public void noCancelPrevious() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Observable.range(1, 5)
+        .concatMap(new Function<Integer, ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> apply(Integer v) throws Exception {
+                return Observable.just(v).doOnDispose(new Action() {
+                    @Override
+                    public void run() throws Exception {
+                        counter.getAndIncrement();
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1, 2, 3, 4, 5);
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatTest.java
index fc2a01619b..d47c684cfc 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatTest.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableConcatTest.java
@@ -28,6 +28,7 @@
 import io.reactivex.Observer;
 import io.reactivex.disposables.*;
 import io.reactivex.exceptions.TestException;
+import io.reactivex.functions.Action;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.functions.Functions;
 import io.reactivex.observers.*;
@@ -1155,4 +1156,42 @@ public void onComplete() {
 
         assertTrue(disposable[0].isDisposed());
     }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void noCancelPreviousArray() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Observable<Integer> source = Observable.just(1).doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        Observable.concatArray(source, source, source, source, source)
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void noCancelPreviousIterable() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Observable<Integer> source = Observable.just(1).doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        Observable.concat(Arrays.asList(source, source, source, source, source))
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableRepeatTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableRepeatTest.java
index 3616ef729f..3afc1f1cc7 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRepeatTest.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRepeatTest.java
@@ -324,4 +324,77 @@ public Object apply(Object w) throws Exception {
         .test()
         .assertFailure(TestException.class, 1, 2, 3);
     }
+
+    @Test
+    public void noCancelPreviousRepeat() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Observable<Integer> source = Observable.just(1).doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.repeat(5)
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRepeatUntil() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Observable<Integer> source = Observable.just(1).doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        final AtomicInteger times = new AtomicInteger();
+
+        source.repeatUntil(new BooleanSupplier() {
+            @Override
+            public boolean getAsBoolean() throws Exception {
+                return times.getAndIncrement() == 4;
+            }
+        })
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRepeatWhen() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        Observable<Integer> source = Observable.just(1).doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        final AtomicInteger times = new AtomicInteger();
+
+        source.repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
+            @Override
+            public ObservableSource<?> apply(Observable<Object> e) throws Exception {
+                return e.takeWhile(new Predicate<Object>() {
+                    @Override
+                    public boolean test(Object v) throws Exception {
+                        return times.getAndIncrement() < 4;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1, 1, 1, 1, 1);
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableRetryTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableRetryTest.java
index 5c1cbe0041..e576479921 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRetryTest.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRetryTest.java
@@ -14,6 +14,7 @@
 package io.reactivex.internal.operators.observable;
 
 import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.*;
@@ -29,6 +30,7 @@
 import io.reactivex.disposables.*;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.*;
+import io.reactivex.internal.functions.Functions;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
 import io.reactivex.observables.GroupedObservable;
 import io.reactivex.observers.*;
@@ -522,11 +524,14 @@ static final class SlowObservable implements ObservableSource<Long> {
         final AtomicInteger active = new AtomicInteger(0), maxActive = new AtomicInteger(0);
         final AtomicInteger nextBeforeFailure;
 
+        final String context;
+
         private final int emitDelay;
 
-        SlowObservable(int emitDelay, int countNext) {
+        SlowObservable(int emitDelay, int countNext, String context) {
             this.emitDelay = emitDelay;
             this.nextBeforeFailure = new AtomicInteger(countNext);
+            this.context = context;
         }
 
         @Override
@@ -542,7 +547,7 @@ public void run() {
             efforts.getAndIncrement();
             active.getAndIncrement();
             maxActive.set(Math.max(active.get(), maxActive.get()));
-            final Thread thread = new Thread() {
+            final Thread thread = new Thread(context) {
                 @Override
                 public void run() {
                     long nr = 0;
@@ -552,7 +557,9 @@ public void run() {
                             if (nextBeforeFailure.getAndDecrement() > 0) {
                                 observer.onNext(nr++);
                             } else {
+                                active.decrementAndGet();
                                 observer.onError(new RuntimeException("expected-failed"));
+                                break;
                             }
                         }
                     } catch (InterruptedException t) {
@@ -613,7 +620,7 @@ public void testUnsubscribeAfterError() {
         Observer<Long> observer = TestHelper.mockObserver();
 
         // Observable that always fails after 100ms
-        SlowObservable so = new SlowObservable(100, 0);
+        SlowObservable so = new SlowObservable(100, 0, "testUnsubscribeAfterError");
         Observable<Long> o = Observable.unsafeCreate(so).retry(5);
 
         AsyncObserver<Long> async = new AsyncObserver<Long>(observer);
@@ -637,7 +644,7 @@ public void testTimeoutWithRetry() {
         Observer<Long> observer = TestHelper.mockObserver();
 
         // Observable that sends every 100ms (timeout fails instead)
-        SlowObservable so = new SlowObservable(100, 10);
+        SlowObservable so = new SlowObservable(100, 10, "testTimeoutWithRetry");
         Observable<Long> o = Observable.unsafeCreate(so).timeout(80, TimeUnit.MILLISECONDS).retry(5);
 
         AsyncObserver<Long> async = new AsyncObserver<Long>(observer);
@@ -931,4 +938,197 @@ public ObservableSource<Object> apply(Throwable ignore) throws Exception {
       assertFalse(subject.hasObservers());
     }
 
+    @Test
+    public void noCancelPreviousRetry() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Observable<Integer> source = Observable.defer(new Callable<ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> call() throws Exception {
+                if (times.getAndIncrement() < 4) {
+                    return Observable.error(new TestException());
+                }
+                return Observable.just(1);
+            }
+        })
+        .doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retry(5)
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRetryWhile() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Observable<Integer> source = Observable.defer(new Callable<ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> call() throws Exception {
+                if (times.getAndIncrement() < 4) {
+                    return Observable.error(new TestException());
+                }
+                return Observable.just(1);
+            }
+        })
+        .doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retry(5, Functions.alwaysTrue())
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRetryWhile2() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Observable<Integer> source = Observable.defer(new Callable<ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> call() throws Exception {
+                if (times.getAndIncrement() < 4) {
+                    return Observable.error(new TestException());
+                }
+                return Observable.just(1);
+            }
+        })
+        .doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retry(new BiPredicate<Integer, Throwable>() {
+            @Override
+            public boolean test(Integer a, Throwable b) throws Exception {
+                return a < 5;
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRetryUntil() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Observable<Integer> source = Observable.defer(new Callable<ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> call() throws Exception {
+                if (times.getAndIncrement() < 4) {
+                    return Observable.error(new TestException());
+                }
+                return Observable.just(1);
+            }
+        })
+        .doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retryUntil(new BooleanSupplier() {
+            @Override
+            public boolean getAsBoolean() throws Exception {
+                return false;
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRepeatWhen() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Observable<Integer> source = Observable.defer(new Callable<ObservableSource<Integer>>() {
+            @Override
+            public ObservableSource<Integer> call() throws Exception {
+                if (times.get() < 4) {
+                    return Observable.error(new TestException());
+                }
+                return Observable.just(1);
+            }
+        }).doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
+            @Override
+            public ObservableSource<?> apply(Observable<Throwable> e) throws Exception {
+                return e.takeWhile(new Predicate<Object>() {
+                    @Override
+                    public boolean test(Object v) throws Exception {
+                        return times.getAndIncrement() < 4;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(0, counter.get());
+    }
+
+    @Test
+    public void noCancelPreviousRepeatWhen2() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        final AtomicInteger times = new AtomicInteger();
+
+        Observable<Integer> source = Observable.<Integer>error(new TestException()).doOnDispose(new Action() {
+            @Override
+            public void run() throws Exception {
+                counter.getAndIncrement();
+            }
+        });
+
+        source.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
+            @Override
+            public ObservableSource<?> apply(Observable<Throwable> e) throws Exception {
+                return e.takeWhile(new Predicate<Object>() {
+                    @Override
+                    public boolean test(Object v) throws Exception {
+                        return times.getAndIncrement() < 4;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult();
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableRetryWithPredicateTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableRetryWithPredicateTest.java
index 72981f705e..7b0322ab87 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRetryWithPredicateTest.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRetryWithPredicateTest.java
@@ -229,7 +229,7 @@ public void testUnsubscribeAfterError() {
         Observer<Long> observer = TestHelper.mockObserver();
 
         // Observable that always fails after 100ms
-        ObservableRetryTest.SlowObservable so = new ObservableRetryTest.SlowObservable(100, 0);
+        ObservableRetryTest.SlowObservable so = new ObservableRetryTest.SlowObservable(100, 0, "testUnsubscribeAfterError");
         Observable<Long> o = Observable
                 .unsafeCreate(so)
                 .retry(retry5);
@@ -255,7 +255,7 @@ public void testTimeoutWithRetry() {
         Observer<Long> observer = TestHelper.mockObserver();
 
         // Observable that sends every 100ms (timeout fails instead)
-        ObservableRetryTest.SlowObservable so = new ObservableRetryTest.SlowObservable(100, 10);
+        ObservableRetryTest.SlowObservable so = new ObservableRetryTest.SlowObservable(100, 10, "testTimeoutWithRetry");
         Observable<Long> o = Observable
                 .unsafeCreate(so)
                 .timeout(80, TimeUnit.MILLISECONDS)
diff --git a/src/test/java/io/reactivex/internal/subscriptions/SubscriptionArbiterTest.java b/src/test/java/io/reactivex/internal/subscriptions/SubscriptionArbiterTest.java
index 02d7d30c0a..426d1779e1 100644
--- a/src/test/java/io/reactivex/internal/subscriptions/SubscriptionArbiterTest.java
+++ b/src/test/java/io/reactivex/internal/subscriptions/SubscriptionArbiterTest.java
@@ -26,7 +26,7 @@ public class SubscriptionArbiterTest {
 
     @Test
     public void setSubscriptionMissed() {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
 
         sa.getAndIncrement();
 
@@ -45,7 +45,7 @@ public void setSubscriptionMissed() {
 
     @Test
     public void invalidDeferredRequest() {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
 
         List<Throwable> errors = TestHelper.trackPluginErrors();
         try {
@@ -59,7 +59,7 @@ public void invalidDeferredRequest() {
 
     @Test
     public void unbounded() {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
 
         sa.request(Long.MAX_VALUE);
 
@@ -86,7 +86,7 @@ public void unbounded() {
 
     @Test
     public void cancelled() {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
         sa.cancelled = true;
 
         BooleanSubscription bs1 = new BooleanSubscription();
@@ -102,7 +102,7 @@ public void cancelled() {
 
     @Test
     public void drainUnbounded() {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
 
         sa.getAndIncrement();
 
@@ -113,7 +113,7 @@ public void drainUnbounded() {
 
     @Test
     public void drainMissedRequested() {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
 
         sa.getAndIncrement();
 
@@ -128,7 +128,7 @@ public void drainMissedRequested() {
 
     @Test
     public void drainMissedRequestedProduced() {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
 
         sa.getAndIncrement();
 
@@ -147,7 +147,7 @@ public void drainMissedRequestedProduced() {
     public void drainMissedRequestedMoreProduced() {
         List<Throwable> errors = TestHelper.trackPluginErrors();
         try {
-            SubscriptionArbiter sa = new SubscriptionArbiter();
+            SubscriptionArbiter sa = new SubscriptionArbiter(true);
 
             sa.getAndIncrement();
 
@@ -169,7 +169,7 @@ public void drainMissedRequestedMoreProduced() {
 
     @Test
     public void missedSubscriptionNoPrior() {
-        SubscriptionArbiter sa = new SubscriptionArbiter();
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
 
         sa.getAndIncrement();
 
@@ -181,4 +181,130 @@ public void missedSubscriptionNoPrior() {
 
         assertSame(bs1, sa.actual);
     }
+
+    @Test
+    public void noCancelFastPath() {
+        SubscriptionArbiter sa = new SubscriptionArbiter(false);
+
+        BooleanSubscription bs1 = new BooleanSubscription();
+        BooleanSubscription bs2 = new BooleanSubscription();
+
+        sa.setSubscription(bs1);
+        sa.setSubscription(bs2);
+
+        assertFalse(bs1.isCancelled());
+        assertFalse(bs2.isCancelled());
+    }
+
+    @Test
+    public void cancelFastPath() {
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
+
+        BooleanSubscription bs1 = new BooleanSubscription();
+        BooleanSubscription bs2 = new BooleanSubscription();
+
+        sa.setSubscription(bs1);
+        sa.setSubscription(bs2);
+
+        assertTrue(bs1.isCancelled());
+        assertFalse(bs2.isCancelled());
+    }
+
+    @Test
+    public void noCancelSlowPathReplace() {
+        SubscriptionArbiter sa = new SubscriptionArbiter(false);
+
+        BooleanSubscription bs1 = new BooleanSubscription();
+        BooleanSubscription bs2 = new BooleanSubscription();
+        BooleanSubscription bs3 = new BooleanSubscription();
+
+        sa.setSubscription(bs1);
+
+        sa.getAndIncrement();
+
+        sa.setSubscription(bs2);
+        sa.setSubscription(bs3);
+
+        sa.drainLoop();
+
+        assertFalse(bs1.isCancelled());
+        assertFalse(bs2.isCancelled());
+        assertFalse(bs3.isCancelled());
+    }
+
+    @Test
+    public void cancelSlowPathReplace() {
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
+
+        BooleanSubscription bs1 = new BooleanSubscription();
+        BooleanSubscription bs2 = new BooleanSubscription();
+        BooleanSubscription bs3 = new BooleanSubscription();
+
+        sa.setSubscription(bs1);
+
+        sa.getAndIncrement();
+
+        sa.setSubscription(bs2);
+        sa.setSubscription(bs3);
+
+        sa.drainLoop();
+
+        assertTrue(bs1.isCancelled());
+        assertTrue(bs2.isCancelled());
+        assertFalse(bs3.isCancelled());
+    }
+
+    @Test
+    public void noCancelSlowPath() {
+        SubscriptionArbiter sa = new SubscriptionArbiter(false);
+
+        BooleanSubscription bs1 = new BooleanSubscription();
+        BooleanSubscription bs2 = new BooleanSubscription();
+
+        sa.setSubscription(bs1);
+
+        sa.getAndIncrement();
+
+        sa.setSubscription(bs2);
+
+        sa.drainLoop();
+
+        assertFalse(bs1.isCancelled());
+        assertFalse(bs2.isCancelled());
+    }
+
+    @Test
+    public void cancelSlowPath() {
+        SubscriptionArbiter sa = new SubscriptionArbiter(true);
+
+        BooleanSubscription bs1 = new BooleanSubscription();
+        BooleanSubscription bs2 = new BooleanSubscription();
+
+        sa.setSubscription(bs1);
+
+        sa.getAndIncrement();
+
+        sa.setSubscription(bs2);
+
+        sa.drainLoop();
+
+        assertTrue(bs1.isCancelled());
+        assertFalse(bs2.isCancelled());
+    }
+
+    @Test
+    public void moreProducedViolationFastPath() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            SubscriptionArbiter sa = new SubscriptionArbiter(true);
+
+            sa.produced(2);
+
+            assertEquals(0, sa.requested);
+
+            TestHelper.assertError(errors, 0, IllegalStateException.class, "More produced than requested: -2");
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }