diff --git a/src/main/java/rx/internal/operators/OnSubscribeRedo.java b/src/main/java/rx/internal/operators/OnSubscribeRedo.java
index 1ba5d1f281..1431d4581c 100644
--- a/src/main/java/rx/internal/operators/OnSubscribeRedo.java
+++ b/src/main/java/rx/internal/operators/OnSubscribeRedo.java
@@ -35,7 +35,6 @@
 
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Notification;
 import rx.Observable;
@@ -47,13 +46,15 @@
 import rx.functions.Action0;
 import rx.functions.Func1;
 import rx.functions.Func2;
+import rx.internal.producers.ProducerArbiter;
+import rx.observers.Subscribers;
 import rx.schedulers.Schedulers;
-import rx.subjects.PublishSubject;
+import rx.subjects.BehaviorSubject;
 import rx.subscriptions.SerialSubscription;
 
 public final class OnSubscribeRedo<T> implements OnSubscribe<T> {
 
-    static final Func1<Observable<? extends Notification<?>>, Observable<?>> REDO_INIFINITE = new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
+    static final Func1<Observable<? extends Notification<?>>, Observable<?>> REDO_INFINITE = new Func1<Observable<? extends Notification<?>>, Observable<?>>() {
         @Override
         public Observable<?> call(Observable<? extends Notification<?>> ts) {
             return ts.map(new Func1<Notification<?>, Notification<?>>() {
@@ -120,7 +121,7 @@ public Notification<Integer> call(Notification<Integer> n, Notification<?> term)
     }
 
     public static <T> Observable<T> retry(Observable<T> source) {
-        return retry(source, REDO_INIFINITE);
+       return retry(source, REDO_INFINITE);
     }
 
     public static <T> Observable<T> retry(Observable<T> source, final long count) {
@@ -144,7 +145,7 @@ public static <T> Observable<T> repeat(Observable<T> source) {
     }
 
     public static <T> Observable<T> repeat(Observable<T> source, Scheduler scheduler) {
-        return repeat(source, REDO_INIFINITE, scheduler);
+        return repeat(source, REDO_INFINITE, scheduler);
     }
 
     public static <T> Observable<T> repeat(Observable<T> source, final long count) {
@@ -172,10 +173,10 @@ public static <T> Observable<T> redo(Observable<T> source, Func1<? super Observa
         return create(new OnSubscribeRedo<T>(source, notificationHandler, false, false, scheduler));
     }
 
-    private Observable<T> source;
+    private final Observable<T> source;
     private final Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> controlHandlerFunction;
-    private boolean stopOnComplete;
-    private boolean stopOnError;
+    private final boolean stopOnComplete;
+    private final boolean stopOnError;
     private final Scheduler scheduler;
 
     private OnSubscribeRedo(Observable<T> source, Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> f, boolean stopOnComplete, boolean stopOnError,
@@ -189,11 +190,12 @@ private OnSubscribeRedo(Observable<T> source, Func1<? super Observable<? extends
 
     @Override
     public void call(final Subscriber<? super T> child) {
-        final AtomicBoolean isLocked = new AtomicBoolean(true);
+        
+        // when true is a marker to say we are ready to resubscribe to source
         final AtomicBoolean resumeBoundary = new AtomicBoolean(true);
+        
         // incremented when requests are made, decremented when requests are fulfilled
         final AtomicLong consumerCapacity = new AtomicLong(0l);
-        final AtomicReference<Producer> currentProducer = new AtomicReference<Producer>();
 
         final Scheduler.Worker worker = scheduler.createWorker();
         child.add(worker);
@@ -201,8 +203,18 @@ public void call(final Subscriber<? super T> child) {
         final SerialSubscription sourceSubscriptions = new SerialSubscription();
         child.add(sourceSubscriptions);
 
-        final PublishSubject<Notification<?>> terminals = PublishSubject.create();
-
+        // use a subject to receive terminals (onCompleted and onError signals) from 
+        // the source observable. We use a BehaviorSubject because subscribeToSource 
+        // may emit a terminal before the restarts observable (transformed terminals) 
+        // is subscribed
+        final BehaviorSubject<Notification<?>> terminals = BehaviorSubject.create();
+        final Subscriber<Notification<?>> dummySubscriber = Subscribers.empty();
+        // subscribe immediately so the last emission will be replayed to the next 
+        // subscriber (which is the one we care about)
+        terminals.subscribe(dummySubscriber);
+
+        final ProducerArbiter arbiter = new ProducerArbiter();
+        
         final Action0 subscribeToSource = new Action0() {
             @Override
             public void call() {
@@ -212,11 +224,11 @@ public void call() {
 
                 Subscriber<T> terminalDelegatingSubscriber = new Subscriber<T>() {
                     boolean done;
+
                     @Override
                     public void onCompleted() {
                         if (!done) {
                             done = true;
-                            currentProducer.set(null);
                             unsubscribe();
                             terminals.onNext(Notification.createOnCompleted());
                         }
@@ -226,7 +238,6 @@ public void onCompleted() {
                     public void onError(Throwable e) {
                         if (!done) {
                             done = true;
-                            currentProducer.set(null);
                             unsubscribe();
                             terminals.onNext(Notification.createOnError(e));
                         }
@@ -235,20 +246,30 @@ public void onError(Throwable e) {
                     @Override
                     public void onNext(T v) {
                         if (!done) {
-                            if (consumerCapacity.get() != Long.MAX_VALUE) {
-                                consumerCapacity.decrementAndGet();
-                            }
                             child.onNext(v);
+                            decrementConsumerCapacity();
+                            arbiter.produced(1);
+                        }
+                    }
+
+                    private void decrementConsumerCapacity() {
+                        // use a CAS loop because we don't want to decrement the
+                        // value if it is Long.MAX_VALUE
+                        while (true) {
+                            long cc = consumerCapacity.get();
+                            if (cc != Long.MAX_VALUE) {
+                                if (consumerCapacity.compareAndSet(cc, cc - 1)) {
+                                    break;
+                                }
+                            } else {
+                                break;
+                            }
                         }
                     }
 
                     @Override
                     public void setProducer(Producer producer) {
-                        currentProducer.set(producer);
-                        long c = consumerCapacity.get();
-                        if (c > 0) {
-                            producer.request(c);
-                        }
+                        arbiter.setProducer(producer);
                     }
                 };
                 // new subscription each time so if it unsubscribes itself it does not prevent retries
@@ -278,12 +299,11 @@ public void onError(Throwable e) {
 
                             @Override
                             public void onNext(Notification<?> t) {
-                                if (t.isOnCompleted() && stopOnComplete)
-                                    child.onCompleted();
-                                else if (t.isOnError() && stopOnError)
-                                    child.onError(t.getThrowable());
-                                else {
-                                    isLocked.set(false);
+                                if (t.isOnCompleted() && stopOnComplete) {
+                                    filteredTerminals.onCompleted();
+                                } else if (t.isOnError() && stopOnError) {
+                                    filteredTerminals.onError(t.getThrowable());
+                                } else {
                                     filteredTerminals.onNext(t);
                                 }
                             }
@@ -313,10 +333,15 @@ public void onError(Throwable e) {
 
                     @Override
                     public void onNext(Object t) {
-                        if (!isLocked.get() && !child.isUnsubscribed()) {
+                        if (!child.isUnsubscribed()) {
+                            // perform a best endeavours check on consumerCapacity 
+                            // with the intent of only resubscribing immediately 
+                            // if there is outstanding capacity
                             if (consumerCapacity.get() > 0) {
                                 worker.schedule(subscribeToSource);
                             } else {
+                                // set this to true so that on next request
+                                // subscribeToSource will be scheduled
                                 resumeBoundary.compareAndSet(false, true);
                             }
                         }
@@ -334,13 +359,11 @@ public void setProducer(Producer producer) {
 
             @Override
             public void request(final long n) {
-                long c = BackpressureUtils.getAndAddRequest(consumerCapacity, n);
-                Producer producer = currentProducer.get();
-                if (producer != null) {
-                    producer.request(n);
-                } else
-                if (c == 0 && resumeBoundary.compareAndSet(true, false)) {
-                    worker.schedule(subscribeToSource);
+                if (n > 0) {
+                    BackpressureUtils.getAndAddRequest(consumerCapacity, n);
+                    arbiter.request(n);
+                    if (resumeBoundary.compareAndSet(true, false))
+                        worker.schedule(subscribeToSource);
                 }
             }
         });
diff --git a/src/test/java/rx/internal/operators/OperatorRetryTest.java b/src/test/java/rx/internal/operators/OperatorRetryTest.java
index a5aa9f1c31..146ee3c254 100644
--- a/src/test/java/rx/internal/operators/OperatorRetryTest.java
+++ b/src/test/java/rx/internal/operators/OperatorRetryTest.java
@@ -395,9 +395,13 @@ public static class FuncWithErrors implements Observable.OnSubscribe<String> {
         public void call(final Subscriber<? super String> o) {
             o.setProducer(new Producer() {
                 final AtomicLong req = new AtomicLong();
+                // 0 = not set, 1 = fast path, 2 = backpressure
+                final AtomicInteger path = new AtomicInteger(0);
+                volatile boolean done = false;
+                
                 @Override
                 public void request(long n) {
-                    if (n == Long.MAX_VALUE) {
+                    if (n == Long.MAX_VALUE && path.compareAndSet(0, 1)) {
                         o.onNext("beginningEveryTime");
                         int i = count.getAndIncrement();
                         if (i < numFailures) {
@@ -408,11 +412,12 @@ public void request(long n) {
                         }
                         return;
                     }
-                    if (n > 0 && req.getAndAdd(n) == 0) {
+                    if (n > 0 && req.getAndAdd(n) == 0 && (path.get() == 2 || path.compareAndSet(0, 2)) && !done) {
                         int i = count.getAndIncrement();
                         if (i < numFailures) {
                             o.onNext("beginningEveryTime");
                             o.onError(new RuntimeException("forced failure: " + (i + 1)));
+                            done = true;
                         } else {
                             do {
                                 if (i == numFailures) {
@@ -421,6 +426,7 @@ public void request(long n) {
                                 if (i > numFailures) {
                                     o.onNext("onSuccessOnly");
                                     o.onCompleted();
+                                    done = true;
                                     break;
                                 }
                                 i = count.getAndIncrement();
@@ -682,91 +688,88 @@ public void testTimeoutWithRetry() {
         assertEquals("Start 6 threads, retry 5 then fail on 6", 6, so.efforts.get());
     }
     
-    @Test(timeout = 15000)
+    @Test//(timeout = 15000)
     public void testRetryWithBackpressure() throws InterruptedException {
-        final int NUM_RETRIES = RxRingBuffer.SIZE * 2;
-        for (int i = 0; i < 400; i++) {
-            @SuppressWarnings("unchecked")
-            Observer<String> observer = mock(Observer.class);
-            Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
-            TestSubscriber<String> ts = new TestSubscriber<String>(observer);
-            origin.retry().observeOn(Schedulers.computation()).unsafeSubscribe(ts);
-            ts.awaitTerminalEvent(5, TimeUnit.SECONDS);
-            
-            InOrder inOrder = inOrder(observer);
-            // should have no errors
-            verify(observer, never()).onError(any(Throwable.class));
-            // should show NUM_RETRIES attempts
-            inOrder.verify(observer, times(NUM_RETRIES + 1)).onNext("beginningEveryTime");
-            // should have a single success
-            inOrder.verify(observer, times(1)).onNext("onSuccessOnly");
-            // should have a single successful onCompleted
-            inOrder.verify(observer, times(1)).onCompleted();
-            inOrder.verifyNoMoreInteractions();
+        final int NUM_LOOPS = 1;
+        for (int j=0;j<NUM_LOOPS;j++) {
+            final int NUM_RETRIES = RxRingBuffer.SIZE * 2;
+            for (int i = 0; i < 400; i++) {
+                @SuppressWarnings("unchecked")
+                Observer<String> observer = mock(Observer.class);
+                Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
+                TestSubscriber<String> ts = new TestSubscriber<String>(observer);
+                origin.retry().observeOn(Schedulers.computation()).unsafeSubscribe(ts);
+                ts.awaitTerminalEvent(5, TimeUnit.SECONDS);
+                
+                InOrder inOrder = inOrder(observer);
+                // should have no errors
+                verify(observer, never()).onError(any(Throwable.class));
+                // should show NUM_RETRIES attempts
+                inOrder.verify(observer, times(NUM_RETRIES + 1)).onNext("beginningEveryTime");
+                // should have a single success
+                inOrder.verify(observer, times(1)).onNext("onSuccessOnly");
+                // should have a single successful onCompleted
+                inOrder.verify(observer, times(1)).onCompleted();
+                inOrder.verifyNoMoreInteractions();
+            }
         }
     }
     
-    @Test(timeout = 15000)
+    @Test//(timeout = 15000)
     public void testRetryWithBackpressureParallel() throws InterruptedException {
+        final int NUM_LOOPS = 1;
         final int NUM_RETRIES = RxRingBuffer.SIZE * 2;
         int ncpu = Runtime.getRuntime().availableProcessors();
         ExecutorService exec = Executors.newFixedThreadPool(Math.max(ncpu / 2, 2));
-        final AtomicInteger timeouts = new AtomicInteger();
-        final Map<Integer, List<String>> data = new ConcurrentHashMap<Integer, List<String>>();
-        final Map<Integer, List<Throwable>> exceptions = new ConcurrentHashMap<Integer, List<Throwable>>();
-        final Map<Integer, Integer> completions = new ConcurrentHashMap<Integer, Integer>();
-        
-        int m = 5000;
-        final CountDownLatch cdl = new CountDownLatch(m);
-        for (int i = 0; i < m; i++) {
-            final int j = i;
-            exec.execute(new Runnable() {
-                @Override
-                public void run() {
-                    final AtomicInteger nexts = new AtomicInteger();
-                    try {
-                        Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
-                        TestSubscriber<String> ts = new TestSubscriber<String>();
-                        origin.retry()
-                        .observeOn(Schedulers.computation()).unsafeSubscribe(ts);
-                        ts.awaitTerminalEvent(2500, TimeUnit.MILLISECONDS);
-                        if (ts.getOnCompletedEvents().size() != 1) {
-                            completions.put(j, ts.getOnCompletedEvents().size());
-                        }
-                        if (ts.getOnErrorEvents().size() != 0) {
-                            exceptions.put(j, ts.getOnErrorEvents());
-                        }
-                        if (ts.getOnNextEvents().size() != NUM_RETRIES + 2) {
-                            data.put(j, ts.getOnNextEvents());
+        try {
+            for (int r = 0; r < NUM_LOOPS; r++) {
+                if (r % 10 == 0) {
+                    System.out.println("testRetryWithBackpressureParallelLoop -> " + r);
+                }
+
+                final AtomicInteger timeouts = new AtomicInteger();
+                final Map<Integer, List<String>> data = new ConcurrentHashMap<Integer, List<String>>();
+
+                int m = 5000;
+                final CountDownLatch cdl = new CountDownLatch(m);
+                for (int i = 0; i < m; i++) {
+                    final int j = i;
+                    exec.execute(new Runnable() {
+                        @Override
+                        public void run() {
+                            final AtomicInteger nexts = new AtomicInteger();
+                            try {
+                                Observable<String> origin = Observable.create(new FuncWithErrors(NUM_RETRIES));
+                                TestSubscriber<String> ts = new TestSubscriber<String>();
+                                origin.retry()
+                                .observeOn(Schedulers.computation()).unsafeSubscribe(ts);
+                                ts.awaitTerminalEvent(2500, TimeUnit.MILLISECONDS);
+                                List<String> onNextEvents = new ArrayList<String>(ts.getOnNextEvents());
+                                if (onNextEvents.size() != NUM_RETRIES + 2) {
+                                    for (Throwable t : ts.getOnErrorEvents()) {
+                                        onNextEvents.add(t.toString());
+                                    }
+                                    for (Object o : ts.getOnCompletedEvents()) {
+                                        onNextEvents.add("onCompleted");
+                                    }
+                                    data.put(j, onNextEvents);
+                                }
+                            } catch (Throwable t) {
+                                timeouts.incrementAndGet();
+                                System.out.println(j + " | " + cdl.getCount() + " !!! " + nexts.get());
+                            }
+                            cdl.countDown();
                         }
-                    } catch (Throwable t) {
-                        timeouts.incrementAndGet();
-                        System.out.println(j + " | " + cdl.getCount() + " !!! " + nexts.get());
-                    }
-                    cdl.countDown();
+                    });
                 }
-            });
-        }
-        exec.shutdown();
-        cdl.await();
-        assertEquals(0, timeouts.get());
-        if (data.size() > 0) {
-            System.out.println(allSequenceFrequency(data));
-        }
-        if (exceptions.size() > 0) {
-            System.out.println(exceptions);
-        }
-        if (completions.size() > 0) {
-            System.out.println(completions);
-        }
-        if (data.size() > 0) {
-            fail("Data content mismatch: " + allSequenceFrequency(data));
-        }
-        if (exceptions.size() > 0) {
-            fail("Exceptions received: " + exceptions);
-        }
-        if (completions.size() > 0) {
-            fail("Multiple completions received: " + completions);
+                cdl.await();
+                assertEquals(0, timeouts.get());
+                if (data.size() > 0) {
+                    fail("Data content mismatch: " + allSequenceFrequency(data));
+                }
+            }
+        } finally {
+            exec.shutdown();
         }
     }
     static <T> StringBuilder allSequenceFrequency(Map<Integer, List<T>> its) {
@@ -783,10 +786,10 @@ static <T> StringBuilder allSequenceFrequency(Map<Integer, List<T>> its) {
     }
     static <T> StringBuilder sequenceFrequency(Iterable<T> it) {
         StringBuilder sb = new StringBuilder();
-        
+
         Object prev = null;
         int cnt = 0;
-        
+
         for (Object curr : it) {
             if (sb.length() > 0) {
                 if (!curr.equals(prev)) {
@@ -805,10 +808,13 @@ static <T> StringBuilder sequenceFrequency(Iterable<T> it) {
             }
             prev = curr;
         }
-        
+        if (cnt > 1) {
+            sb.append(" x ").append(cnt);
+        }
+
         return sb;
     }
-    @Test(timeout = 3000)
+    @Test//(timeout = 3000)
     public void testIssue1900() throws InterruptedException {
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
@@ -849,7 +855,7 @@ public Observable<String> call(GroupedObservable<String, String> t1) {
         inOrder.verify(observer, times(1)).onCompleted();
         inOrder.verifyNoMoreInteractions();
     }
-    @Test(timeout = 3000)
+    @Test//(timeout = 3000)
     public void testIssue1900SourceNotSupportingBackpressure() {
         @SuppressWarnings("unchecked")
         Observer<String> observer = mock(Observer.class);
diff --git a/src/test/java/rx/internal/operators/OperatorRetryWithPredicateTest.java b/src/test/java/rx/internal/operators/OperatorRetryWithPredicateTest.java
index ee4750829a..76461e3ddf 100644
--- a/src/test/java/rx/internal/operators/OperatorRetryWithPredicateTest.java
+++ b/src/test/java/rx/internal/operators/OperatorRetryWithPredicateTest.java
@@ -20,7 +20,10 @@
 import static org.mockito.Mockito.*;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.*;
 
@@ -305,4 +308,56 @@ public Integer call(Integer t1) {
 
         assertEquals(1, value);
     }
+    
+    @Test
+    public void testIssue3008RetryWithPredicate() {
+        final List<Long> list = new CopyOnWriteArrayList<Long>();
+        final AtomicBoolean isFirst = new AtomicBoolean(true);
+        Observable.<Long> just(1L, 2L, 3L).map(new Func1<Long, Long>(){
+            @Override
+            public Long call(Long x) {
+                System.out.println("map " + x);
+                if (x == 2 && isFirst.getAndSet(false)) {
+                    throw new RuntimeException("retryable error");
+                }
+                return x;
+            }})
+        .retry(new Func2<Integer, Throwable, Boolean>() {
+            @Override
+            public Boolean call(Integer t1, Throwable t2) {
+                return true;
+            }})
+        .forEach(new Action1<Long>() {
+
+            @Override
+            public void call(Long t) {
+                System.out.println(t);
+                list.add(t);
+            }});
+        assertEquals(Arrays.asList(1L,1L,2L,3L), list);
+    }
+    
+    @Test
+    public void testIssue3008RetryInfinite() {
+        final List<Long> list = new CopyOnWriteArrayList<Long>();
+        final AtomicBoolean isFirst = new AtomicBoolean(true);
+        Observable.<Long> just(1L, 2L, 3L).map(new Func1<Long, Long>(){
+            @Override
+            public Long call(Long x) {
+                System.out.println("map " + x);
+                if (x == 2 && isFirst.getAndSet(false)) {
+                    throw new RuntimeException("retryable error");
+                }
+                return x;
+            }})
+        .retry()
+        .forEach(new Action1<Long>() {
+
+            @Override
+            public void call(Long t) {
+                System.out.println(t);
+                list.add(t);
+            }});
+        assertEquals(Arrays.asList(1L,1L,2L,3L), list);
+    }
 }