Exception handling using launch instead of relying on async

[carterhudson]

I'm writing a small helper class to bridge some asynchronous functionality via coroutines to some of our Java classes. I'm a little confused about handling errors when using launch coroutines, so I'd like to check my expectations (I've read #61). I've gotten this to work successfully using async instead of launch, but I really would like to understand how to do this using launch.

Given the following:

val someData: SomeData? = null

fun returnSomeJob(): Job {
    ...
    //some other operations here
    ...

    return doAsyncThings()
}

fun doAsyncThings(): Job = launch(UI) {
    try {
        someData = getSomeDataAsync()
    } catch (e: Throwable) {
        //do something with the error locally, re-throw to percolate
        throw e
    }
}

suspend fun getSomeDataAsync(): SomeData = getSomeResponse().await().let{ unwrapOrThrow(it) }

fun unwrapOrThrow(someResponse: SomeResponse): SomeData {
    when(someResponse.isSuccessful()) {
        true -> someResponse.getSomeData()
        false -> throw SomeException()
    }
}

What would I need to do to make the following helper class forward errors through a callback properly?

class CoroutineHelper {
    companion object {
        interface CompletionHandler<T> {
            fun onSuccess(value: T)
            fun onError(error: Throwable)
        }

        @Suppress("UNCHECKED_CAST")
        @JvmStatic
        fun <T> handleCompletion(job: Job, handler: CompletionHandler<T>): Job {
            return launch(UI) {
                when (job) {
                //we care about a returned value
                    is Deferred<*> -> {
                        try {
                            "Expecting a return value...".dlog()
                            (job.await() as T).let { handler.onSuccess(it) }
                        } catch (e: Throwable) {
                            handler.onError(e)
                        }
                    }

                // we do not care about a returned value
                    else -> {
                        try {
                            "Not expecting a return value...".dlog()
                            job.join()
                            handler.onSuccess(Unit as T)
                        } catch (e: Throwable) {
                            handler.onError(e)
                        }
                    }
                }
            }
        }
    }
}

I've tried adding a CoroutineExceptionHandler to various contexts up the chain, but no combination of launch(UI + CoroutineExceptionHandler { ... }) on any or all coroutine builders achieves what I'm after. launch(UI + CoroutineExceptionHandler{ launch(UI) { ... } }) allows me to intercept errors in the first launch, which I can log, but not re-throw in order to percolate. I'd like to bubble them up to catch them in handleCompletion.

[jcornaz]

First, as a general advice, always prefer suspending functions over functions returning Job or Deferred. With suspending function error handling is as simple as standard try-catch-finally.

Then for a usage from Java I would personally convert any Deferred and Job to a CompletableFuture:

So the CoroutinesHelper would be:

object CoroutineHelper {

  @JvmStatic
  fun <T> asFuture(defferred: Deferred<T>): CompletableFuture<T> {
    val result = CompletableFuture<T>()

    defferred.invokeOnCompletion { exception ->
      if (exception != null) {
        result.completeExceptionally(exception)
      } else {
        result.complete(defferred.getCompleted())
      }
    }

    return result
  }

  @JvmStatic
  fun asFuture(job: Job): CompletableFuture<Void> {
    val result = CompletableFuture<Void>()

    job.invokeOnCompletion { exception ->
      if (exception != null) {
        result.completeExceptionally(exception)
      } else {
        result.complete(null)
      }
    }

    return result
  }
}

Then you can use the CompletableFuture as usual in Java (join, get, thenAccept, thenCompose, exceptionally, etc.)

Note that kotlinx-coroutines-jdk8 give you CompletionStage.asDeferred() and CompletionStage.await() extension functions to use CompletableFuture from coroutines.

[carterhudson]

@jcornaz
I do normally prefer suspending functions, but I don't think that interoperates well, or at all, with Java. As for the completable futures, it's my understanding that Java 8 is required. However, the codebase I'm working in doesn't currently have Java 8 support. As for the implementation of the asFuture(...) methods, I was originally using invokeOnCompletion as a sort of callback for the job before I decided I needed to implement a bridging helper. I was motivated to avoid using invokeOnCompletion because of the following documentation:

* **Note**: This function is a part of internal machinery that supports parent-child hierarchies
* and allows for implementation of suspending functions that wait on the Job's state.
* This function should not be used in general application code.
* Implementations of `CompletionHandler` must be fast and _lock-free_.
*/
public fun invokeOnCompletion(onCancelling: Boolean = false, handler: CompletionHandler): DisposableHandle

In general, I'm aware that there are probably better ways to handle coroutine interop between Kotlin and Java, but some of the required tools are unavailable to me in this specific instance. Mostly, I'd like to know how to do the things mentioned in the OP for my own edification.

[jcornaz]

@carterhudson, suspend function may be used from Java. You have to pass one more argument at the end of type Continuation which is basically a callback. However, it wouldn't be my advice to use it as it is quite ugly and error prone.

Your reasons to not use invokeOnCompletion are indeed good. But you may simply write your own:

fun Job.onCompletion(callback: (Throwable?) -> Unit) = launch(Unconfined) {
  join()

  var exception: Throwable? = getCancellationException()

  if (!isCancelled && exception is JobCancellationException) {
    exception = exception.cause // will be null if the job completed normally
  }

  callback(exception)
}

Using CompletableFuture was just an example. If you don't have access to Java8 you can achieve the same with any library providing Futures for Java.

Here is another example using RxJava and kotlinx-coroutines-rx2:

object CoroutineHelper {

  @JvmStatic
  fun <T> asSingle(deferred: Deferred<T>): Single<T> = deferred.asSingle(Unconfined)

  @JvmStatic
  fun asCompletable(job: Job): Completable = job.asCompletable(Unconfied)
}

Of course you could even write your own operators on Deferred and Job to make them handleable from Java, but it would be quite cumbersome and error prone. If you have to use Coroutines from java, my advice would be to wrap Deferred and Job with your preferred Java library for future/promise management. (Instead of CompletableFuture you may pick AsyncTask if you're doing android development, Single and Completable if you like RxJava, otherwise pick any library providing future/promise you like). Then you use the error management of the given library.

Coroutines' benefits relies on the language keyword suspend. Without it (when used from Java) coroutines are pointless IMO, so it's better to bridge to a Java library than using Kotlin coroutines from Java.

But of course from Kotlin using suspend functions is awesome, so just convert as much java code to kotlin as possible 😁

[carterhudson]

I wasn't aware of kotlinx-coroutines-rx2; I'm actually a big fan of reactive extensions. I might pull that in and go down that path.

I think I might be able to get rid of the bridging class by simply returning a Throwable? instead of a Job and keying off the nullability of the error object, or just having the method itself throw and annotate with @Throws. Keeps things more sequential as well and seems obvious now that I think about it 🤔 . Seems I got ahead of myself here. This doesn't actually work. The Java just executes synchronously since coroutines and suspension are virtues of the Kotlin compiler.

As for converting as much Java to Kotlin as possible... if only they'd let me 😁 . Thanks for the info!

I am still rather curious what I'm doing wrong with launch and CoroutineExceptionHandler, though.

[jcornaz]

It looks like you expect join to throw in case of job failure. But it will not.

This is why async should be used when you care about the result (in term of value and/or of success). launch is a fire and forget which should be used only when you don't care about the result value nor the success.

Finally if, despite it is contrary to the philosophy, you really want to handle the exception of a Job when joining it. You have to use getCancellationException and check its type as well as verifying if the job has been cancelled.

Here's an example:

suspend fun Job.joinOrThrow() {
  join()

  var exception: Throwable? = getCancellationExcepion() 

  if (!isCancelled && exception is JobCancellationException) {
    exception = exception.cause // will be null if the job completed normally
  }

  if (exception != null) throw exception
}

Note that launch will always give the exception to the coroutine exception handler (which will be by default delegated to the uncaught exception handler). And async will never do that as await is meant to be called.

[dave08]

@jcornaz, I was also looking for such a thing, I'm not a fan of having

Deffered<Unit>

just for seeing if there was an exception... Wouldn't that be a great addition to the lib? Also don't you mean '!is JobCancellationException'?

[jcornaz]

Also don't you mean '!is JobCancellationException

No i meant is. getCancellationException will return a JobCancellationException if the job has been cancelled or terminated normally. See the doc: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-job/get-cancellation-exception.html

Wouldn't that be a great addition to the lib?

IMO no. In fact I would even advice to not use the joinOrThrow I wrote in my last answer.

This is actually the difference between Deffered<Unit> and Job. They are different and both have there usage. If you want to know if the Job failed, then use Deferred<Unit>, otherwise use Job. This doesn't bother me at all. Either I care about the success and use async, either I don't and use launch.

Also keep in mind that Deffered and Job should be used at top level to explicitly execute thinks concurrently. Avoid writing function returning Deferred or Job, and make your functions suspend instead. Then it will be even simpler and easier to use (just try-catch).

Finally, let's assume we want to add it in the library. How should it behave? If we continue to delegate the exception to the CoroutineExceptionHandler the exception will be handled twice (by the handler and then by the code calling join). But if we don't delegate to the exception handler it means that user would have to call join for each job created, which is a non-sense as launch is meant to be a fire-and-forget.

However I agree it is not intuitive. I felt myself in the trap the first time I used it. May be it should be better documented?

[dave08]

Thanks for the explanation! It is currently very confusing... From what I understand, the best way to handle launch exceptions is either using try catch in launch itself or using a CoroutineExceptionHandler which is like an onError handler in rx and the gain is to not require joining... It's not really fire and forget, but that the calling code can forget it... I always thought this is a subject missing documentation, but mainly in terms of best practices and clear use cases...

Also, funny that after isCancelled is false, the exception should be JobCancellationException? The naming is a bit misleading...

[jcornaz]

the calling code can forget it

Indeed that what is meant by "fire and forget". In practice you may still join, use a CoroutineExceptionHandler, handle the exception inside the launch block or even call a callback! (don't do this).

I always thought this is a subject missing documentation, but mainly in terms of best practices and clear use cases...

Me too. May be we should have a look and propose an improvement of the documentation. May be in the form of a "best practices" document?

Also, funny that after isCancelled is false, the exception should be JobCancellationException? The naming is a bit misleading...

Yes, I agree. Or maybe I just haven't thought about it enough. Because I avoid using it. I stick to make correct choices between using launch and async and I never need to call getCancellationException.

[dave08]

even call a callback! (don't do this).

What do you mean, CoroutineExceptionHandler also to be avoided?

never need to call getCancellationException .

Logging is a common use case, since if something crashes, you want to be notified even if you don't need the result...

[jcornaz]

CoroutineExceptionHandler also to be avoided?

No sorry. I only meant don't call callback from launch. This would be against the whole philosophy and it would kill the point of using coroutines. But attaching a CoroutineExceptionHandler is perfectly fine as long as it done is for good reasons (logging for example).

Just don't do it in order to go back to a callback programming style:

fun notGood(onSuccess: (Int) -> Unit, onError: (Throwable) -> Unit) {
  launch(CoroutineExceptionHandler { _, throwable -> onError(throwable) }) {
    val result = computeAnswerToLifeUniverseAndEverything() // suspend for 7.5 million years and return 42
    onSuccess(result)
  }
}

Logging is a common use case, since if something crashes, you want to be notified even if you don't need the result...

Logging is indeed a common case of a good usage of CoroutineExceptionHandler. No need to use async if your only concern about failure is logging. And no need to call getCancellationException either.

val ExceptionLogger = CoroutineExceptionHandler { context, throwable ->
  println("$throwable happened in $context") // log
}

fun good() {
  launch(ExceptionLogger) {
    val result = computeAnswerToLifeUniverseAndEverything() // suspend for 7.5 million years and return 42
    println(result)
  }
}

However, in our code base we decided to attach the log directly to the default Thread.UncaughtExceptionHandler, so we don't have to manually provide a CoroutineExceptionHandler each time we use launch and we think that all uncaught exception should be logged anyway.