diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCache.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCache.java
index daa9edd533..d2e74aefb8 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCache.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCache.java
@@ -24,23 +24,7 @@
  *
  * @param <T> the source element type
  */
-public final class ObservableCache<T> extends AbstractObservableWithUpstream<T, T>
-implements Observer<T> {
-
-    /**
-     * The subscription to the source should happen at most once.
-     */
-    final AtomicBoolean once;
-
-    /**
-     * The number of items per cached nodes.
-     */
-    final int capacityHint;
-
-    /**
-     * The current known array of observer state to notify.
-     */
-    final AtomicReference<CacheDisposable<T>[]> observers;
+public final class ObservableCache<T> extends AbstractObservableWithUpstream<T, T> {
 
     /**
      * A shared instance of an empty array of observers to avoid creating
@@ -56,61 +40,49 @@ public final class ObservableCache<T> extends AbstractObservableWithUpstream<T,
     static final CacheDisposable[] TERMINATED = new CacheDisposable[0];
 
     /**
-     * The total number of elements in the list available for reads.
-     */
-    volatile long size;
-
-    /**
-     * The starting point of the cached items.
-     */
-    final Node<T> head;
-
-    /**
-     * The current tail of the linked structure holding the items.
-     */
-    Node<T> tail;
-
-    /**
-     * How many items have been put into the tail node so far.
+     * The subscription to the source should happen at most once.
      */
-    int tailOffset;
+    final AtomicBoolean once;
 
     /**
-     * If {@link #observers} is {@link #TERMINATED}, this holds the terminal error if not null.
+     * Responsible caching events from the source and multicasting them to each downstream.
      */
-    Throwable error;
+    final Multicaster<T>  multicaster;
 
     /**
-     * True if the source has terminated.
+     * The first node in a singly linked list. Each node has the capacity to hold a specific number of events, and each
+     * points exclusively to the <em>next</em> node (if present). When a new downstream arrives, the subscription is
+     * initialized with a reference to the "head" node, and any events present in the linked list are replayed. As
+     * events are replayed to the new downstream, its 'node' reference advances through the linked list, discarding each
+     * node reference once all events in that node have been replayed. Consequently, once {@code this} instance goes out
+     * of scope, the prefix of nodes up to the first node that is still being replayed becomes unreachable and eligible
+     * for collection.
      */
-    volatile boolean done;
+    final Node<T> head;
 
     /**
      * Constructs an empty, non-connected cache.
      * @param source the source to subscribe to for the first incoming observer
      * @param capacityHint the number of items expected (reduce allocation frequency)
      */
-    @SuppressWarnings("unchecked")
     public ObservableCache(Observable<T> source, int capacityHint) {
         super(source);
-        this.capacityHint = capacityHint;
         this.once = new AtomicBoolean();
         Node<T> n = new Node<>(capacityHint);
         this.head = n;
-        this.tail = n;
-        this.observers = new AtomicReference<>(EMPTY);
+        this.multicaster = new Multicaster<>(capacityHint, n);
     }
 
     @Override
     protected void subscribeActual(Observer<? super T> t) {
-        CacheDisposable<T> consumer = new CacheDisposable<>(t, this);
+        CacheDisposable<T> consumer = new CacheDisposable<>(t, multicaster, head);
         t.onSubscribe(consumer);
-        add(consumer);
+        multicaster.add(consumer);
 
         if (!once.get() && once.compareAndSet(false, true)) {
-            source.subscribe(this);
+            source.subscribe(multicaster);
         } else {
-            replay(consumer);
+            multicaster.replay(consumer);
         }
     }
 
@@ -127,7 +99,7 @@ protected void subscribeActual(Observer<? super T> t) {
      * @return true if the cache has observers
      */
     /* public */ boolean hasObservers() {
-        return observers.get().length != 0;
+        return multicaster.get().length != 0;
     }
 
     /**
@@ -135,194 +107,238 @@ protected void subscribeActual(Observer<? super T> t) {
      * @return the number of currently cached event count
      */
     /* public */ long cachedEventCount() {
-        return size;
+        return multicaster.size;
     }
 
-    /**
-     * Atomically adds the consumer to the {@link #observers} copy-on-write array
-     * if the source has not yet terminated.
-     * @param consumer the consumer to add
-     */
-    void add(CacheDisposable<T> consumer) {
-        for (;;) {
-            CacheDisposable<T>[] current = observers.get();
-            if (current == TERMINATED) {
-                return;
-            }
-            int n = current.length;
+    static final class Multicaster<T> extends AtomicReference<CacheDisposable<T>[]> implements Observer<T> {
 
-            @SuppressWarnings("unchecked")
-            CacheDisposable<T>[] next = new CacheDisposable[n + 1];
-            System.arraycopy(current, 0, next, 0, n);
-            next[n] = consumer;
+        /**
+         * The number of items per cached nodes.
+         */
+        final int capacityHint;
 
-            if (observers.compareAndSet(current, next)) {
-                return;
-            }
-        }
-    }
+        /**
+         * The total number of elements in the list available for reads.
+         */
+        volatile long size;
 
-    /**
-     * Atomically removes the consumer from the {@link #observers} copy-on-write array.
-     * @param consumer the consumer to remove
-     */
-    @SuppressWarnings("unchecked")
-    void remove(CacheDisposable<T> consumer) {
-        for (;;) {
-            CacheDisposable<T>[] current = observers.get();
-            int n = current.length;
-            if (n == 0) {
-                return;
-            }
+        /**
+         * The current tail of the linked structure holding the items.
+         */
+        Node<T> tail;
 
-            int j = -1;
-            for (int i = 0; i < n; i++) {
-                if (current[i] == consumer) {
-                    j = i;
-                    break;
-                }
-            }
+        /**
+         * How many items have been put into the tail node so far.
+         */
+        int tailOffset;
 
-            if (j < 0) {
-                return;
-            }
-            CacheDisposable<T>[] next;
+        /**
+         * If the observers are {@link #TERMINATED}, this holds the terminal error if not null.
+         */
+        Throwable error;
 
-            if (n == 1) {
-                next = EMPTY;
-            } else {
-                next = new CacheDisposable[n - 1];
-                System.arraycopy(current, 0, next, 0, j);
-                System.arraycopy(current, j + 1, next, j, n - j - 1);
-            }
+        /**
+         * True if the source has terminated.
+         */
+        volatile boolean done;
 
-            if (observers.compareAndSet(current, next)) {
-                return;
-            }
+        @SuppressWarnings("unchecked")
+        Multicaster(int capacityHint, final Node<T> head) {
+            super(EMPTY);
+            this.tail = head;
+            this.capacityHint = capacityHint;
         }
-    }
 
-    /**
-     * Replays the contents of this cache to the given consumer based on its
-     * current state and number of items requested by it.
-     * @param consumer the consumer to continue replaying items to
-     */
-    void replay(CacheDisposable<T> consumer) {
-        // make sure there is only one replay going on at a time
-        if (consumer.getAndIncrement() != 0) {
-            return;
-        }
+        /**
+         * Atomically adds the consumer to the observers copy-on-write array
+         * if the source has not yet terminated.
+         * @param consumer the consumer to add
+         */
+        void add(CacheDisposable<T> consumer) {
+            for (;;) {
+                CacheDisposable<T>[] current = get();
+                if (current == TERMINATED) {
+                    return;
+                }
+                int n = current.length;
 
-        // see if there were more replay request in the meantime
-        int missed = 1;
-        // read out state into locals upfront to avoid being re-read due to volatile reads
-        long index = consumer.index;
-        int offset = consumer.offset;
-        Node<T> node = consumer.node;
-        Observer<? super T> downstream = consumer.downstream;
-        int capacity = capacityHint;
-
-        for (;;) {
-            // if the consumer got disposed, clear the node and quit
-            if (consumer.disposed) {
-                consumer.node = null;
-                return;
+                @SuppressWarnings("unchecked")
+                CacheDisposable<T>[] next = new CacheDisposable[n + 1];
+                System.arraycopy(current, 0, next, 0, n);
+                next[n] = consumer;
+
+                if (compareAndSet(current, next)) {
+                    return;
+                }
             }
+        }
+
+        /**
+         * Atomically removes the consumer from the observers copy-on-write array.
+         * @param consumer the consumer to remove
+         */
+        @SuppressWarnings("unchecked")
+        void remove(CacheDisposable<T> consumer) {
+            for (;;) {
+                CacheDisposable<T>[] current = get();
+                int n = current.length;
+                if (n == 0) {
+                    return;
+                }
+
+                int j = -1;
+                for (int i = 0; i < n; i++) {
+                    if (current[i] == consumer) {
+                        j = i;
+                        break;
+                    }
+                }
+
+                if (j < 0) {
+                    return;
+                }
+                CacheDisposable<T>[] next;
 
-            // first see if the source has terminated, read order matters!
-            boolean sourceDone = done;
-            // and if the number of items is the same as this consumer has received
-            boolean empty = size == index;
-
-            // if the source is done and we have all items so far, terminate the consumer
-            if (sourceDone && empty) {
-                // release the node object to avoid leaks through retained consumers
-                consumer.node = null;
-                // if error is not null then the source failed
-                Throwable ex = error;
-                if (ex != null) {
-                    downstream.onError(ex);
+                if (n == 1) {
+                    next = EMPTY;
                 } else {
-                    downstream.onComplete();
+                    next = new CacheDisposable[n - 1];
+                    System.arraycopy(current, 0, next, 0, j);
+                    System.arraycopy(current, j + 1, next, j, n - j - 1);
                 }
+
+                if (compareAndSet(current, next)) {
+                    return;
+                }
+            }
+        }
+
+        /**
+         * Replays the contents of this cache to the given consumer based on its
+         * current state and number of items requested by it.
+         * @param consumer the consumer to continue replaying items to
+         */
+        void replay(CacheDisposable<T> consumer) {
+            // make sure there is only one replay going on at a time
+            if (consumer.getAndIncrement() != 0) {
                 return;
             }
 
-            // there are still items not sent to the consumer
-            if (!empty) {
-             // if the offset in the current node has reached the node capacity
-                if (offset == capacity) {
-                    // switch to the subsequent node
-                    node = node.next;
-                    // reset the in-node offset
-                    offset = 0;
+            // see if there were more replay request in the meantime
+            int missed = 1;
+            // read out state into locals upfront to avoid being re-read due to volatile reads
+            long index = consumer.index;
+            int offset = consumer.offset;
+            Node<T> node = consumer.node;
+            Observer<? super T> downstream = consumer.downstream;
+            int capacity = capacityHint;
+
+            for (;;) {
+                // if the consumer got disposed, clear the node and quit
+                if (consumer.disposed) {
+                    consumer.node = null;
+                    return;
                 }
 
-                // emit the cached item
-                downstream.onNext(node.values[offset]);
-
-                // move the node offset forward
-                offset++;
-                // move the total consumed item count forward
-                index++;
+                // first see if the source has terminated, read order matters!
+                boolean sourceDone = done;
+                // and if the number of items is the same as this consumer has received
+                boolean empty = size == index;
+
+                // if the source is done and we have all items so far, terminate the consumer
+                if (sourceDone && empty) {
+                    // release the node object to avoid leaks through retained consumers
+                    consumer.node = null;
+                    // if error is not null then the source failed
+                    Throwable ex = error;
+                    if (ex != null) {
+                        downstream.onError(ex);
+                    } else {
+                        downstream.onComplete();
+                    }
+                    return;
+                }
 
-                // retry for the next item/terminal event if any
-                continue;
-            }
+                // there are still items not sent to the consumer
+                if (!empty) {
+                 // if the offset in the current node has reached the node capacity
+                    if (offset == capacity) {
+                        // switch to the subsequent node
+                        node = node.next;
+                        // reset the in-node offset
+                        offset = 0;
+                    }
+
+                    // emit the cached item
+                    downstream.onNext(node.values[offset]);
+
+                    // move the node offset forward
+                    offset++;
+                    // move the total consumed item count forward
+                    index++;
+
+                    // retry for the next item/terminal event if any
+                    continue;
+                }
 
-            // commit the changed references back
-            consumer.index = index;
-            consumer.offset = offset;
-            consumer.node = node;
-            // release the changes and see if there were more replay request in the meantime
-            missed = consumer.addAndGet(-missed);
-            if (missed == 0) {
-                break;
+                // commit the changed references back
+                consumer.index = index;
+                consumer.offset = offset;
+                consumer.node = node;
+                // release the changes and see if there were more replay request in the meantime
+                missed = consumer.addAndGet(-missed);
+                if (missed == 0) {
+                    break;
+                }
             }
         }
-    }
 
-    @Override
-    public void onSubscribe(Disposable d) {
-        // we can't do much with the upstream disposable
-    }
-
-    @Override
-    public void onNext(T t) {
-        int tailOffset = this.tailOffset;
-        // if the current tail node is full, create a fresh node
-        if (tailOffset == capacityHint) {
-            Node<T> n = new Node<>(tailOffset);
-            n.values[0] = t;
-            this.tailOffset = 1;
-            tail.next = n;
-            tail = n;
-        } else {
-            tail.values[tailOffset] = t;
-            this.tailOffset = tailOffset + 1;
+        @Override
+        public void onSubscribe(Disposable d) {
+            // we can't do much with the upstream disposable
         }
-        size++;
-        for (CacheDisposable<T> consumer : observers.get()) {
-            replay(consumer);
+
+        @Override
+        public void onNext(T t) {
+            int tailOffset = this.tailOffset;
+            // if the current tail node is full, create a fresh node
+            if (tailOffset == capacityHint) {
+                Node<T> n = new Node<>(tailOffset);
+                n.values[0] = t;
+                this.tailOffset = 1;
+                tail.next = n;
+                tail = n;
+            } else {
+                tail.values[tailOffset] = t;
+                this.tailOffset = tailOffset + 1;
+            }
+            size++;
+            for (CacheDisposable<T> consumer : get()) {
+                replay(consumer);
+            }
         }
-    }
 
-    @SuppressWarnings("unchecked")
-    @Override
-    public void onError(Throwable t) {
-        error = t;
-        done = true;
-        for (CacheDisposable<T> consumer : observers.getAndSet(TERMINATED)) {
-            replay(consumer);
+        @SuppressWarnings("unchecked")
+        @Override
+        public void onError(Throwable t) {
+            error = t;
+            done = true;
+            // No additional events will arrive, so now we can clear the 'tail' reference
+            tail = null;
+            for (CacheDisposable<T> consumer : getAndSet(TERMINATED)) {
+                replay(consumer);
+            }
         }
-    }
 
-    @SuppressWarnings("unchecked")
-    @Override
-    public void onComplete() {
-        done = true;
-        for (CacheDisposable<T> consumer : observers.getAndSet(TERMINATED)) {
-            replay(consumer);
+        @SuppressWarnings("unchecked")
+        @Override
+        public void onComplete() {
+            done = true;
+            // No additional events will arrive, so now we can clear the 'tail' reference
+            tail = null;
+            for (CacheDisposable<T> consumer : getAndSet(TERMINATED)) {
+                replay(consumer);
+            }
         }
     }
 
@@ -338,7 +354,7 @@ static final class CacheDisposable<T> extends AtomicInteger
 
         final Observer<? super T> downstream;
 
-        final ObservableCache<T> parent;
+        final Multicaster<T> parent;
 
         Node<T> node;
 
@@ -353,11 +369,12 @@ static final class CacheDisposable<T> extends AtomicInteger
          * the parent cache object.
          * @param downstream the actual consumer
          * @param parent the parent that holds onto the cached items
+         * @param head the first node in the linked list
          */
-        CacheDisposable(Observer<? super T> downstream, ObservableCache<T> parent) {
+        CacheDisposable(Observer<? super T> downstream, Multicaster<T> parent, Node<T> head) {
             this.downstream = downstream;
             this.parent = parent;
-            this.node = parent.head;
+            this.node = head;
         }
 
         @Override
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCacheTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCacheTest.java
index 7f47ec95d8..74d17c062b 100644
--- a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCacheTest.java
+++ b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableCacheTest.java
@@ -16,10 +16,15 @@
 import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+import java.lang.management.MemoryUsage;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
+import io.reactivex.rxjava3.observables.ConnectableObservable;
 import org.junit.Test;
 
 import io.reactivex.rxjava3.core.*;
@@ -355,4 +360,52 @@ public void addRemoveRace() {
             );
         }
     }
+
+    @Test
+    public void valuesAreReclaimable() throws Exception {
+        ConnectableObservable<byte[]> source =
+                Observable.range(0, 200)
+                        .map($ -> new byte[1024 * 1024])
+                        .publish();
+
+        System.out.println("Bounded Replay Leak check: Wait before GC");
+        Thread.sleep(1000);
+
+        System.out.println("Bounded Replay Leak check: GC");
+        System.gc();
+
+        Thread.sleep(500);
+
+        final MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean();
+        MemoryUsage memHeap = memoryMXBean.getHeapMemoryUsage();
+        long initial = memHeap.getUsed();
+
+        System.out.printf("Bounded Replay Leak check: Starting: %.3f MB%n", initial / 1024.0 / 1024.0);
+
+        final AtomicLong after = new AtomicLong();
+
+        source.cache().lastElement().subscribe(new Consumer<byte[]>() {
+            @Override
+            public void accept(byte[] v) throws Exception {
+                System.out.println("Bounded Replay Leak check: Wait before GC 2");
+                Thread.sleep(1000);
+
+                System.out.println("Bounded Replay Leak check:  GC 2");
+                System.gc();
+
+                Thread.sleep(500);
+
+                after.set(memoryMXBean.getHeapMemoryUsage().getUsed());
+            }
+        });
+
+        source.connect();
+
+        System.out.printf("Bounded Replay Leak check: After: %.3f MB%n", after.get() / 1024.0 / 1024.0);
+
+        if (initial + 100 * 1024 * 1024 < after.get()) {
+            fail("Bounded Replay Leak check: Memory leak detected: " + (initial / 1024.0 / 1024.0)
+                    + " -> " + after.get() / 1024.0 / 1024.0);
+        }
+    }
 }