Here is a quick fix (not elegant):

```java
    volatile Scheduler.Inner innerScheduler = null;

    /**
     * Try draining the queue on the given scheduler.
     * The method ensures that only one thread is actively draining the
     * queue on the given scheduler.
     * 
     * @param scheduler
     *            the scheduler where the draining should happen
     * @param cs
     *            the composite subscription to track the schedule
     */
    public void tryDrainAsync(Scheduler scheduler, final CompositeSubscription cs) {
        if (cs.isUnsubscribed() || wip.incrementAndGet() > 1) {
            return;
        }
        if (innerScheduler == null) {
            // add tracking subscription only if schedule is run to avoid overfilling cs
            final MultipleAssignmentSubscription mas = new MultipleAssignmentSubscription();
            cs.add(mas);
            mas.set(scheduler.schedule(new Action1<Scheduler.Inner>() {
                @Override
                public void call(Scheduler.Inner o) {
                    innerScheduler = o;
                    if (!cs.isUnsubscribed()) {
                        do {
                            queue.poll().call();
                        } while (wip.decrementAndGet() > 0 && !cs.isUnsubscribed());
                    }
                }
            }));
        } else {
            innerScheduler.schedule(new Action1<Scheduler.Inner>() {
                @Override
                public void call(Scheduler.Inner o) {
                    if (!cs.isUnsubscribed()) {
                        do {
                            queue.poll().call();
                        } while (wip.decrementAndGet() > 0 && !cs.isUnsubscribed());
                    }
                }
            });
        }
    }
```