Resilience4j incompatible with rxjava3

[bgreen1005]

Resilience4j version:
2.2.0

Java version:
17

We're using resilience4j in our spring boot app as the circuit breaker, and using rxjava as our reactive library. We're currently trying to update rxjava2 to rxjava3. It looks to us that the circuit breaker is no longer falling back to the fallback method when an error is thrown with rxjava3. I've created a simplified project (attached) which demonstrates the problem, but here's a brief overview of the demo:

The service just returns "Success" if throwException is false, and returns an Observable.error if throwException is true.

@Component
public class MyService {

    @CircuitBreaker(name = "Test", fallbackMethod = "fallback")
    public Single<String> getResult(boolean throwException) {
        return Single.fromObservable(process(throwException));
    }

    Observable<String> process(boolean throwException) {
        if (throwException) {
            return Observable.error(new IllegalArgumentException("Test"));
        }
        return Observable.just("Success");
    }

    private Single<String> fallback(boolean throwException, Throwable throwable) {
        return Single.just("Fallback");
    }

}

Our expectation is that when throwException is true, the fallback method should get called, and the response we get should be Single.just("Fallback"). The behaviour of the circuit breaker and method is verified with the tests provided:

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = {Application.class})
public class MyServiceTest {

    @Autowired
    private MyService myService;

    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;

    @AfterEach
    void cleanUp() {
        circuitBreakerRegistry.circuitBreaker("Test").reset();
    }

    @Test
    void getResult_returnsFallback_whenThrowExceptionTrue() {
        assertThat(myService.getResult(true).blockingGet()).isEqualTo("Fallback");
    }
    
    @Test
    void getResult_causesCircuitBreakerToRegisterFailedCall_whenThrowExceptionTrue() {
        myService.getResult(true).blockingGet();

        assertThat(circuitBreakerRegistry.circuitBreaker("Test").getMetrics().getNumberOfFailedCalls()).isEqualTo(1);
    }

    @Test
    void getResult_returnsSuccess_whenThrowExceptionFalse() {
        assertThat(myService.getResult(false).blockingGet()).isEqualTo("Success");
    }

    @Test
    void getResult_returnsFallback_whenCircuitBreakerOpen() {
        circuitBreakerRegistry.circuitBreaker("Test").transitionToOpenState();

        assertThat(myService.getResult(false).blockingGet()).isEqualTo("Fallback");
    }
}

When we're using rxjava2, these tests pass as expected. However, when we change our dependencies and imports to rxjava3, the getResult_returnsFallback_whenThrowExceptionTrue test fails as the IllegalArgumentException gets thrown, and isn't handled by the fallback method. The getResult_causesCircuitBreakerToRegisterFailedCall_whenThrowExceptionTrue test also fails, as the circuit breaker has not recorded a failed call.

The build.gradle shows our dependencies:

plugins {
    id "org.springframework.boot" version "2.6.6"
    id 'io.spring.dependency-management' version '1.0.11.RELEASE'
}

apply plugin: 'java'
apply plugin: 'idea'


group = 'org.example'
version = '1.0-SNAPSHOT'

repositories {
    mavenCentral()
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.boot:spring-boot-starter-aop'

    implementation 'io.github.resilience4j:resilience4j-spring-boot2:2.2.0'
    implementation 'io.github.resilience4j:resilience4j-rxjava2:2.2.0'

    implementation 'io.reactivex:rxjava-reactive-streams:1.2.1'

    testImplementation platform('org.junit:junit-bom:5.9.1')
    testImplementation 'org.junit.jupiter:junit-jupiter'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
}

test {
    useJUnitPlatform()
}

and when we go from rx java 2 to rx java 3, we swap the io.github.resilience4j:resilience4j-rxjava2:2.2.0 for io.github.resilience4j:resilience4j-rxjava3:2.2.0 and fix the imports in MyService.

Is there anything we're missing to make this work? Or is this a bug?

It's worth noting that we've tried spring boot 3 too, and the behaviour is identical.

resilience4j-rxjava.zip

[RobWin]

Hi,
since I don't have much time at the moment.
Would it be possible for you to debug it?

[bgreen1005]

I've just had a look, and it looks like there is a class called RxJava2CircuitBreakerAspectExtwhich handles circuit breakers for rxjava2. I think we need a corresponding class for rxjava3. There's a few other RxJava2 classes that we will need equivalents for. Do you want me to pick this up and raise a PR?

[RobWin]

Yes, actually I think we never implemented a RxJava3 Aspect Extension :/

Same here: https://github.com/resilience4j/resilience4j/tree/master/resilience4j-spring/src/main/java/io/github/resilience4j/fallback

We just have a https://github.com/resilience4j/resilience4j/blob/master/resilience4j-spring/src/main/java/io/github/resilience4j/fallback/RxJava2FallbackDecorator.java

[bgreen1005]

I'll raise a PR :) Thanks.

[bgreen1005]

Do you want me to leave support for rx java 2 or remove it? It looks like that's been EOL for a few years now.

[RobWin]

Yes, would make sense. Thank you

[bgreen1005]

I've decided to leave rx2 there for now, as it's still a valid option. It's a pretty big PR but I think it's fairly well tested. Will get it raised tomorrow.

[bgreen1005]

Pull request raised :) It's a biggie, but hopefully nothing too controversial.

[bgreen1005]

@RobWin thanks for merging. Just so I can set some expectations, when do you think we'll be getting the next release?

[bgreen1005]

@RobWin I notice that Sonar isn't very happy. Seems to be complaining about duplicated code since my PR. Is that going to block a release?

[bgreen1005]

@RobWin Sorry to ask again, do you have any idea when this will be released? It's a bit of a blocker for us at the moment.

Also, shall we close this issue, or will we close it after it gets released?

[RobWin]

I usually close it when it's released.
I don't have much time at the moment. Too much work at my company. I try to release it as soon as possible.

But since it's an extension system, anyone can potentially provide AspectExt and FallbackDecorator to resilience4j by providing a separate Spring Boot extension for any new ReturnType.
Then it would not be necessary for me to maintain and release all of these extensions.

You could basically move everything into a new module/library as a resilience4j rxjava3 springboot extension.

But as said I try to release as soon as possible.

[bgreen1005]

Thanks @RobWin , that was going to be my other suggestion if you were too busy - Adding those AspectExt and FallbackDecorator to our classpath until the release is done.

[RobWin]

Yes. that should be possible. At least that was the design idea.
Never really tested it, since no one provided external extensions. :P

[bgreen1005]

I've just tried this out - I've added the RxJava3CircuitBreakerAspectExt and RxJava3FallbackDecorator as Components in our project, and it's working for what we need right now. I'll delete when you do the release, but it should unblock us (and more importantly stop me from nagging you!).

Thanks again!

[RobWin]

Perfect,
thank you for the feedback