Summary

Driver version: 4.5.0

Following up from JAVA-4304, it seems that Observable.map does not take into account whether the inner function threw an exception when calling onComplete.

So, when an exception occurs, MapObservable first calls onError (which is fine) and then onComplete on the the next Observer in the chain, which is against the reactive streams spec.

How to Reproduce

The following code will throw an IllegalStateException in ToSingleObservablePublisher with message "onComplete called after the Observer has already completed or errored."

val future = SingleObservable[Int](1).map( { x =>
  throw new RuntimeException("Some exception")
} ).toSingle().toFuture()

Additional Background

I've managed to circumvent the bug by making a patched version of the Observable.map method that simply does not call onComplete after an exception has occured in the inner function. My code is really bad and Scala 3, though, so sharing it won't help much.  But the fix seems easy enough.