RxJava flatmap crashing even though onError is present

Question

I am trying to debug a crash on Android that occurs when a device does not have a network connection. The crash only occurs sometimes (~50% of the time) and I am not sure why. If anyone can either explain what is going on or give a solution I will be very grateful.

I am using a flat map to fetch images in parallel and I subscribe with an onError() block. Despite this the app still crashes.

Here is the code (I labelled the line it crashes on with "Crash here:"):

fun downloadImages() {
    var progress = 0
    var maxProgress = 0

    downloadModel.postValue(
            Resource.inProgress(getString(R.string.downloading_offline_data)))

    disposables.add(Observable
            .create<Pair<String, File>> { emitter ->
                val files = readFileDirs()

                for (file in files) {
                  val urlList = readUrlListFrom(file)
                  maxProgress += urlList.size

                  for (url in urlList) {
                    emitter.onNext(url to file)
                  }
                }
                emitter.onComplete()
            }
            .flatMap { (url, file) ->
                downloadImage(url, file)
                        .subscribeOn(Schedulers.computation())
                        .andThen(Observable.fromCallable {
                            ++progress
                        })
            }
            .doOnDispose {
                downloadModel.postValue(Resource.idle())
            }
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                Log.d(TAG, "Done downloading $it")

                if (it % 100 == 0) {
                    refreshCacheSize()
                }

                downloadModel.postValue(Resource.inProgress(
                        getString(R.string.downloading_offline_data),
                        it,
                        maxProgress
                ))
            }, { error ->
                val errorId = when (error) {
                    is SocketTimeoutException ->
                        NETWORK_TIMEOUT_ERROR
                    is InterruptedIOException -> {
                        // do nothing
                        return@subscribe
                    }
                    is UnknownHostException ->
                        NETWORK_ERROR
                    is IOException ->
                        UNKNOWN_ERROR
                    is UnsupportedServerVersionException ->
                        UNSUPPORTED_VERSION_ERROR
                    else -> {
                        UNKNOWN_ERROR
                    }
                }
                downloadModel.postValue(Resource.error(errorId))
            }, {
                refreshCacheSize()

                downloadModel.postValue(Resource.success())
            })
    )
}

private fun downloadImage(url: String, outFile: File): Completable {
    val outFileTmp = File(outFile.canonicalPath + ".tmp")
    return Completable
            .create { emitter ->
                val request = Request.Builder()
                        .url(url)
                        .build()
                try {
Crash here:          val response = Client.get().newCall(request).execute()
                    val sink: BufferedSink
                    try {
                        Log.d(TAG, "Writing to file")
                        val body = response.body()
                        if (body != null) {
                            sink = Okio.buffer(Okio.sink(outFileTmp))
                            sink.writeAll(body.source())
                            sink.close()
                        } else {
                            if (!emitter.isDisposed) {
                                emitter.onError(DownloadImageException("Body was null."))
                            }
                        }
                    } finally {
                        response.body()?.close()
                    }

                    outFileTmp.renameTo(outFile)

                    emitter.onComplete()
                } catch (e: Exception) {
                    Log.e(TAG, "Could not save splash", e)
                    emitter.tryOnError(e)
                }
            }
            .doOnDispose {
                outFileTmp.delete()
            }
            .doOnError {
                outFileTmp.delete()
            }
}

Stack trace:

2018-10-08 02:50:07.055 21984-22263/com.ggstudios.lolcatalyst E/AndroidRuntime: FATAL EXCEPTION: RxComputationThreadPool-2
    Process: com.ggstudios.lolcatalyst, PID: 21984
    java.lang.Throwable: Unable to resolve host "mywebsite.com": No address associated with hostname
        at java.net.Inet6AddressImpl.lookupHostByName(Inet6AddressImpl.java:157)
        at java.net.Inet6AddressImpl.lookupAllHostAddr(Inet6AddressImpl.java:105)
        at java.net.InetAddress.getAllByName(InetAddress.java:1154)
        at okhttp3.Dns$1.lookup(Dns.java:40)
        at okhttp3.internal.connection.RouteSelector.resetNextInetSocketAddress(RouteSelector.java:185)
        at okhttp3.internal.connection.RouteSelector.nextProxy(RouteSelector.java:149)
        at okhttp3.internal.connection.RouteSelector.next(RouteSelector.java:84)
        at okhttp3.internal.connection.StreamAllocation.findConnection(StreamAllocation.java:214)
        at okhttp3.internal.connection.StreamAllocation.findHealthyConnection(StreamAllocation.java:135)
        at okhttp3.internal.connection.StreamAllocation.newStream(StreamAllocation.java:114)
        at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:42)
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
        at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:93)
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
        at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93)
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
        at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:126)
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147)
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121)
        at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:200)
        at okhttp3.RealCall.execute(RealCall.java:77)
        at com.ggstudios.lolcatalyst.offline.OfflineDownloadViewModel$downloadImage$1.subscribe(OfflineDownloadViewModel.kt:372)
        at io.reactivex.internal.operators.completable.CompletableCreate.subscribeActual(CompletableCreate.java:39)
        at io.reactivex.Completable.subscribe(Completable.java:1918)
        at hu.akarnokd.rxjava2.debug.CompletableOnAssembly.subscribeActual(CompletableOnAssembly.java:39)
        at io.reactivex.Completable.subscribe(Completable.java:1918)
        at io.reactivex.internal.operators.completable.CompletableSubscribeOn$SubscribeOnObserver.run(CompletableSubscribeOn.java:64)
        at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38)
        at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:301)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)
        at java.lang.Thread.run(Thread.java:764)
     Caused by: java.lang.Throwable: android_getaddrinfo failed: EAI_NODATA (No address associated with hostname)
        at libcore.io.Linux.android_getaddrinfo(Native Method)
        at libcore.io.BlockGuardOs.android_getaddrinfo(BlockGuardOs.java:172)
        at java.net.Inet6AddressImpl.lookupHostByName(Inet6AddressImpl.java:137)
        at java.net.Inet6AddressImpl.lookupAllHostAddr(Inet6AddressImpl.java:105) 
        at java.net.InetAddress.getAllByName(InetAddress.java:1154) 
        at okhttp3.Dns$1.lookup(Dns.java:40) 
        at okhttp3.internal.connection.RouteSelector.resetNextInetSocketAddress(RouteSelector.java:185) 
        at okhttp3.internal.connection.RouteSelector.nextProxy(RouteSelector.java:149) 
        at okhttp3.internal.connection.RouteSelector.next(RouteSelector.java:84) 
        at okhttp3.internal.connection.StreamAllocation.findConnection(StreamAllocation.java:214) 
        at okhttp3.internal.connection.StreamAllocation.findHealthyConnection(StreamAllocation.java:135) 
        at okhttp3.internal.connection.StreamAllocation.newStream(StreamAllocation.java:114) 
        at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:42) 
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147) 
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121) 
        at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:93) 
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147) 
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121) 
        at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93) 
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147) 
        at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:126) 
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:147) 
        at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:121) 
        at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:200) 
        at okhttp3.RealCall.execute(RealCall.java:77) 
        at com.ggstudios.lolcatalyst.offline.OfflineDownloadViewModel$downloadImage$1.subscribe(OfflineDownloadViewModel.kt:372) 
        at io.reactivex.internal.operators.completable.CompletableCreate.subscribeActual(CompletableCreate.java:39) 
        at io.reactivex.Completable.subscribe(Completable.java:1918) 
        at hu.akarnokd.rxjava2.debug.CompletableOnAssembly.subscribeActual(CompletableOnAssembly.java:39) 
        at io.reactivex.Completable.subscribe(Completable.java:1918) 
        at io.reactivex.internal.operators.completable.CompletableSubscribeOn$SubscribeOnObserver.run(CompletableSubscribeOn.java:64) 
        at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38) 
        at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26) 
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:301) 
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167) 
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641) 
        at java.lang.Thread.run(Thread.java:764)

See this. https://stackoverflow.com/questions/43135211/android-unable-to-handle-unknownhostexception-with-retrofit-2-okhttp3-rxandr — Naveen, Oct 08 '18 at 10:35
Possible duplicate of https://stackoverflow.com/questions/52468708/handling-error-rxjava-android-with-kotlin/52468824#52468824 — Ahmed Ashraf, Oct 08 '18 at 12:05
@Ahmed Ashraf Gamal not a duplicate of that one since I am passing in a lambda to handle errors already. — idunnololz, Oct 08 '18 at 17:19
Oh I see, must be the "tryOnError" then, it should be "onError" as far as I know. — Ahmed Ashraf, Oct 08 '18 at 17:28

score 1 · Accepted Answer · answered Oct 11 '18 at 10:32

I figured out both the cause and some possible solutions.

Let's start with the cause first.

My code can be simplified like this: Observable #1 starts off n other Observables that execute on other threads via flatMap().

For simplicity let's assume that all of the Observables execute on their own thread. Then when an error is thrown by one observable, the error is propagated to all of the other Observables and dispose them.

The issue is that this propagation is not synchronized (the reasoning appears to be for performance reasons). Because of this tryOnError() can pass the isDisposed() check, then be disposed, and then fire the error. Which explains why the exception still occurs even though tryOnError() is used.

Some solutions

I ultimately came up with two main solutions after some research and discussion.

One way is to convert all errors into results in the Observables via onErrorReturn(). This solution works but might not be great if you want to halt execution as soon as an error is hit.

The other way to is to register a global error handler and just ignore these errors.

I went with the latter solution however I am not fully satisfied. I really wish there was a way to register a local error handler. Eg. have a method on Observables to ignore errors emitted after disposal of an Observable.

Oh well.

RxJava flatmap crashing even though onError is present

1 Answers1