Infinite never failing hot flux

Question

I'm using webflux for handling my http request. As an side effect of the processing I want to add record to the database but I do not want to stop processing of user request to achieve that.

Somewhere in main application flow.

@GetMapping
Flux getHandler(){
  return doStuff().doOnNext(data -> dataStore.store(data));
}

In different class I have

class DataStore {
private static final Logger LOGGER = LoggerFactory.getLogger(DataStore.class);

private DataRepository repository;
private Scheduler scheduler;
private Sinks.Many sink;

public DataStore(DataRepository repository, Scheduler scheduler)
  this.repository = repository;
  this.scheduler = scheduler; //will be boundedElastic in production
  this.sink = Sinks.many().replay().limit(1000); //buffer size
  //build hot flux
  this.sink.asFlux()
       .map(data -> repository.save(data))
       // retry strategy for random issues with DB connection
       .retryWhen(Retry.backoff(maxRetry, backoffDuration)
                       .doBeforeRetry(signal -> LOGGER.warn("Retrying to save, attempt {}", signal.totalRetries())))
       // give up on saving this item, drop it, try with another one, reset backoff strategy in the meantime
       .onErrorContinue(Exceptions::isRetryExhausted, (e, o) -> LOGGER.error("Dropping data")) 
       .subscribeOn(scheduler, true)
       .subscribe(
                        data-> LOGGER.info("Data {} saved.", data),
                        error -> LOGGER.error("Fatal error. Terminating store flux.", error)
                );
}

public void store(Data data) {
   sink.tryEmitNext(data);
}

But when writing tests for it I have noticed that if backoff reaches it limit flux instead of doping the data and continuing will just stop.

    @BeforeEach
    public void setup() {
        repository = mock(DataRepository.class);
        dataStore = new DataStore(repository, Schedulers.immediate()); //maxRetry = 4, backoffDuration = Duration.ofMillis(1)
    }

    @Test
    public void test() throws Exception {
        //given
        when(repository.save(any()))
                .thenThrow(new RuntimeException("fail")) // normal store
                .thenThrow(new RuntimeException("fail")) // first retry
                .thenThrow(new RuntimeException("fail")) // second retry
                .thenThrow(new RuntimeException("fail")) // third retry
                .thenThrow(new RuntimeException("fail")) // fourth retry -> should drop data("One")
                .thenAnswer(invocation -> invocation.getArgument(0)) //store data("Two")
                .thenAnswer(invocation -> invocation.getArgument(0));//store data("Three")

        //when
        searchStore.store(data("One")); //exhaust 5 retries
        searchStore.store(data("Two")); //successful store
        searchStore.store(data("Three")); //successful store

        //then
        Thread.sleep(2000); //overkill sleep
        verify(repository, times(7)).save(any()); //assertion fails. data two and three was not saved. 
    }

When running this test my assertion fails and in the logs I can see only

Retrying to save, attempt 0
Retrying to save, attempt 1
Retrying to save, attempt 2
Retrying to save, attempt 3
Dropping data

And there is no info of successful processing of data Two and Three.

I do not want to retry indefinitely, because I assume that DB connection may fail from time to time and I do not want to have buffer overflow. I know that I can achieve similar flow without flux (use queue etc.), but the build in retry with backoff is very tempting.

How I can drop error from the flux as onErrorContinue does not seam to be working?

user_551406 · Answer

General note - the code in the above question isn't used in a reactive context, and therefore this answer suggestions options that would be completely wrong if using Webflux or in a similar reactive environment.

Firstly, note that onErrorContinue() is almost certainly not what you want - not just in this situation, but generally. It's a specialist operator that almost certainly doesn't quite do what you think it does.

Usually, I'd balk at this line:

.map(data -> repository.save(data))

...as it implies your repository isn't reactive, so you're blocking in a reactive chain - a complete no-no. In this case because you're using it purely for the convenience of the retry semantics it's not going to cause issues, but bear in mind most people used to seeing reactive code will get scared when they see stuff like this!

If you're able to use a reactive repository, that would be better, and then you'd expect to see something like this:

.flatMap(data -> repository.save(data))

...implying that the save() method is returning a non-blocking Mono rather than a plain value after blocking. The norm with retrying would then be to retry on the inner publisher, resuming on an empty publisher if retries are exhausted:

.flatMap(data -> repository.save(data)
        .retryWhen(Retry.backoff(maxRetry, backoffDuration)
                        .doBeforeRetry(signal -> LOGGER.warn("Retrying to save, attempt {}", signal.totalRetries())))
        .onErrorResume(Exceptions::isRetryExhausted, e -> Mono.empty()) 
)

If you're not able or willing to use a reactive repository, then in this case you can still achieve the above by wrapping repository.save(data) as Mono.just(repository.save(data)) - but again, that's a bit of a code smell, and completely forbidden in a standard reactive chain, as you're making something "look" reactive when it's not.

Infinite never failing hot flux

1 Answers