The problem about CurrentThreadScheduler is here. When we reach the `onCompleted`, an synchronized Observable has not return the subscription yet. In Observable.java, every subscribe will be wrapped by a SafeObservableSubscription and SafeObserver. If the observer has some error, SafeObserver will unsubscribe the SafeObservableSubscription. But now SafeObservableSubscription has not be set the real Subscription yet. So now when we is in the `onCompleted` here, we has no way to know if unsubscribe has happens. We will continue to call `scheduler.schedule`. 

Next time, when we enter `onCompleted` here again, we still do not know if unsubscribe has happens. So we will not be able to terminate it even if the observer has some error.

```java
                SafeObservableSubscription subscription = new SafeObservableSubscription();
                subscription.wrap(onSubscribeFunction.onSubscribe(new SafeObserver<T>(subscription, observer)));
                return hook.onSubscribeReturn(this, subscription);
```