RxJava Single - Getting a memory leak, how to correctly unsubscribe?

I'm using RxJava's Single.fromCallable() to wrap around a third party library that makes an API call. I was testing different states on the call, success, failed, low network, no network.

But on the no network test I ran into a memory leak for the first time ever using RxJava. I spent the last hour combing through the code and trying to narrow down the leak with the LeakCanary library.

I figured out it was coming from subscribing to the Single.fromCallable().

Single.fromCallable(() -> {
       return remoteRepository.makeTransaction(signedTransaction).execute();
})
       .subscribeOn(Schedulers.newThread())
       .observeOn(AndroidSchedulers.mainThread())
       .subscribe(txHash -> {
              Log.d(TAG, "makeTransaction: " + txHash);
                    
                   
       }, err -> {
             Log.e(TAG, "makeTransaction: ", err);
                    
       });

Once I remove the

.subscribe(txHash -> { ... });

It no longer leaks.

I've tried googling stackoverflow with RxJava Single unsubscribe and I'm getting answers saying that you don't need to unsubscribe from Single

https://stackoverflow.com/a/43332198/11110509.

But if I don't I'll be getting memory leaks.

I've tried to unsubscribe by making the Single call an instance variable in my ViewModel:

Disposable mDisposable;

mDisposable = Single.fromCallable(() -> {...}).subscribe(txHash -> {..});

and unsubscribing it in the Fragment's onDestroy() method so it will unsubscribe if the user exits the screen before the call is finished.

@Override
    public void onDestroy() {
        super.onDestroy();
        mViewModel.getDisposable().dispose();
    }

But it's still leaking. I'm not sure if this is the correct way to unsubscribe or maybe I'm doing something else incorrectly.

How can I correctly unsubscribe from the Single Callable?

Edit:_________________________________

How I'm recreating the issue:

Screen one is launches screen two. The call is performed instantly on the creation of screen two. Since I'm testing it with no network, the query is continuing to perform on screen two until it times out. So closing the screen two before the call finishes causes the leak.

It doesn't seem to be a context leak because I've removed tried testing it by removing all the methods inside the .subscribe():

Single.fromCallable(() -> {
           
            return remoteRepository.makeTransaction(signedTransaction).execute();
})
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(txHash-> {
              //Removed all methods here.
              //Still leaks.

         }, err -> {

                    
         });

But when I remove:

.subscribe(txHash-> {
                   
}, err -> {  
              
});

it no longer leaks.

LeakCanary logs:

┬───
    │ GC Root: Java local variable
    │
    ├─ java.lang.Thread thread
    │    Leaking: UNKNOWN
    │    Retaining 2.4 kB in 81 objects
    │    Thread name: 'RxCachedThreadScheduler-2'
    │    ↓ Thread.<Java Local>
    │             ~~~~~~~~~~~~
    ╰→ com.dave.testapp.ui.send.SendViewModel instance
    ​     Leaking: YES (ObjectWatcher was watching this because com.dave.testapp.ui.send.SendViewModel received
    ​     ViewModel#onCleared() callback)
    ​     Retaining 588 B in 19 objects
    ​     key = 6828ea76-a75c-448b-8278-d0e0bb0229c8
    ​     watchDurationMillis = 10324
    ​     retainedDurationMillis = 5321
    ​     baseApplication instance of com.dave.testapp.BaseApplication
    
    METADATA
    
    Build.VERSION.SDK_INT: 27
    Build.MANUFACTURER: Google
    LeakCanary version: 2.7
    App process name: com.dave.testapp

One easy mistake to be made in Java is forgetting about the implicit reference to the outer class instance whenever you instantiate an anonymous class in a non-static context.

For example:

Single.fromCallable(() -> {
   // some logic
   return 5;
});

Is actually the same as:

Single.fromCallable(new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    // some logic
                    return 5;
                }
            });

So what you did is instantiate a new instance of the anonymous class which implements the Callable interface.

Now, let's put this in some context.

Let's assume we have this inside a service class:

class SomeService {
    int aNumber = 5;
    Single<Integer> doLogic() {
        return Single.fromCallable(() -> {
            // using "aNumber" here implicates Service.this.aNumber
            // e.g. writing "return 5 + aNumber" is actually the same as
            return 5 + SomeService.this.aNumber;
        });
    }
}

Most of the time this isn't an issue because the outer classes outlive the short-lived objects instantiated inside the methods. However, if your instantiated object can outlive the outer object (in your case the Single is still running even after the ViewModel is out of scope), the whole outer object (in your case, the ViewModel) remains in memory as the Callable still has the implicit reference to it.

There are many ways how to get rid of this unwanted reference - the easiest being instantiating the object in a static context where you capture only what you really need (instead of the whole "outer this").

class SomeService {
    int aNumber = 5;
    
    static Callable staticCallableThatCapturesOnlyParameters(int param) {
        return () -> {
            // outer this is not available here
            return 5 + param; // param is captured through the function args
        };
    }
    Single<Integer> doLogic() {
        return Single.fromCallable(staticCallableThatCapturesOnlyParameters(aNumber));
    }
}

Another approach would be to simply avoid the anonymous object altogether and use static inner classes, but that quickly bloats code.