Observable#first causes retryWhen to trigger retry without an exception

[alexwen]

Noticed this behavior using rx-netty as internally it does a take(1) internally:

    AtomicLong inc = new AtomicLong(0);
    Observable.OnSubscribe<Long> onSubscribe = subscriber -> {
        final long emit = inc.getAndIncrement();
        LOGGER.info("Emitting: {}", emit);
        subscriber.onNext(emit);
        subscriber.onCompleted();
    };

    LOGGER.info("DONE: {}", Observable.create(onSubscribe)
                    .retryWhen(attempt -> attempt.zipWith(Observable.range(1, 4), (n, i) -> i))
                    .toBlocking()
                    .first()
    );

The test output is:
Emitting: 0
Emitting: 1
DONE: 0

As you can see the observable is subscribed to twice. This is because take, in combination with single, causes the retryWhen producer to request another item.

[benjchristensen]

Manually merged in #1793

Thank you @alexwen for submitting a bug report and then this fix!