diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishMulticast.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishMulticast.java
index 928c0ffbd7..73b5aa44b6 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishMulticast.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishMulticast.java
@@ -137,7 +137,7 @@ static final class MulticastProcessor<T> extends Flowable<T> implements Subscrib
         final AtomicReference<MulticastSubscription<T>[]> subscribers;
 
         final int prefetch;
-        
+
         final int limit;
 
         final boolean delayError;
@@ -150,7 +150,7 @@ static final class MulticastProcessor<T> extends Flowable<T> implements Subscrib
 
         volatile boolean done;
         Throwable error;
-        
+
         int consumed;
 
         @SuppressWarnings("unchecked")
@@ -319,11 +319,11 @@ void drain() {
             int missed = 1;
 
             SimpleQueue<T> q = queue;
-            
+
             int upstreamConsumed = consumed;
             int localLimit = limit;
             boolean canRequest = sourceMode != QueueSubscription.SYNC;
-            
+
             for (;;) {
                 MulticastSubscription<T>[] array = subscribers.get();
 
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 e0738b951b..b074e10179 100644
--- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java
+++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTimed.java
@@ -174,6 +174,7 @@ public void onComplete() {
         public void dispose() {
             worker.dispose();
             DisposableHelper.dispose(timer);
+            s.cancel();
         }
 
         @Override
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableTimeoutTimed.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableTimeoutTimed.java
index e5a4b6b5f6..0c62048555 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableTimeoutTimed.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableTimeoutTimed.java
@@ -174,6 +174,7 @@ public void onComplete() {
         public void dispose() {
             worker.dispose();
             DisposableHelper.dispose(this);
+            s.dispose();
         }
 
         @Override
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishFunctionTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishFunctionTest.java
index b404f3ddc8..eef409d31e 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishFunctionTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishFunctionTest.java
@@ -473,7 +473,7 @@ public Publisher<Integer> apply(Flowable<Integer> v) throws Exception {
                             public boolean test(Integer w) throws Exception {
                                 return w % 2 == 0;
                             }
-                        }), 
+                        }),
                         v.filter(new Predicate<Integer>() {
                             @Override
                             public boolean test(Integer w) throws Exception {
@@ -500,7 +500,7 @@ public Publisher<Integer> apply(Flowable<Integer> v) throws Exception {
                             public boolean test(Integer w) throws Exception {
                                 return w % 2 == 0;
                             }
-                        }), 
+                        }),
                         v.filter(new Predicate<Integer>() {
                             @Override
                             public boolean test(Integer w) throws Exception {
@@ -528,7 +528,7 @@ public Publisher<Integer> apply(Flowable<Integer> v) throws Exception {
                             public boolean test(Integer w) throws Exception {
                                 return w % 2 == 0;
                             }
-                        }), 
+                        }),
                         v.filter(new Predicate<Integer>() {
                             @Override
                             public boolean test(Integer w) throws Exception {
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTests.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTests.java
index 64dc5c912b..f51d1669e3 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTests.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutTests.java
@@ -483,4 +483,39 @@ protected void subscribeActual(Subscriber<? super Integer> observer) {
         }
     }
 
+
+    @Test
+    public void timedTake() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = ps.timeout(1, TimeUnit.DAYS)
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasSubscribers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasSubscribers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void timedFallbackTake() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = ps.timeout(1, TimeUnit.DAYS, Flowable.just(2))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasSubscribers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasSubscribers());
+
+        to.assertResult(1);
+    }
+
 }
\ No newline at end of file
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutWithSelectorTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutWithSelectorTest.java
index f08b438647..f0f3772dc8 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutWithSelectorTest.java
+++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableTimeoutWithSelectorTest.java
@@ -13,7 +13,7 @@
 
 package io.reactivex.internal.operators.flowable;
 
-import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
@@ -509,4 +509,40 @@ protected void subscribeActual(Subscriber<? super Integer> observer) {
         .test()
         .assertResult(1);
     }
+
+    @Test
+    public void selectorTake() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = ps
+        .timeout(Functions.justFunction(Flowable.never()))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasSubscribers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasSubscribers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void selectorFallbackTake() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = ps
+        .timeout(Functions.justFunction(Flowable.never()), Flowable.just(2))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasSubscribers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasSubscribers());
+
+        to.assertResult(1);
+    }
 }
\ No newline at end of file
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutTests.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutTests.java
index 20fd73d63b..03b2a45e47 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutTests.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutTests.java
@@ -481,4 +481,38 @@ protected void subscribeActual(Observer<? super Integer> observer) {
             RxJavaPlugins.reset();
         }
     }
+
+    @Test
+    public void timedTake() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        TestObserver<Integer> to = ps.timeout(1, TimeUnit.DAYS)
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasObservers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasObservers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void timedFallbackTake() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        TestObserver<Integer> to = ps.timeout(1, TimeUnit.DAYS, Observable.just(2))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasObservers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasObservers());
+
+        to.assertResult(1);
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutWithSelectorTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutWithSelectorTest.java
index 90bbeb2ace..c99c1cf0d3 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutWithSelectorTest.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableTimeoutWithSelectorTest.java
@@ -13,7 +13,7 @@
 
 package io.reactivex.internal.operators.observable;
 
-import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
@@ -510,4 +510,40 @@ protected void subscribeActual(Observer<? super Integer> observer) {
         .test()
         .assertResult(1);
     }
+
+    @Test
+    public void selectorTake() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        TestObserver<Integer> to = ps
+        .timeout(Functions.justFunction(Observable.never()))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasObservers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasObservers());
+
+        to.assertResult(1);
+    }
+
+    @Test
+    public void selectorFallbackTake() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        TestObserver<Integer> to = ps
+        .timeout(Functions.justFunction(Observable.never()), Observable.just(2))
+        .take(1)
+        .test();
+
+        assertTrue(ps.hasObservers());
+
+        ps.onNext(1);
+
+        assertFalse(ps.hasObservers());
+
+        to.assertResult(1);
+    }
 }