Hello good reactor folks - I trying to write some reactive code (surprising eh?) and have hit a slight snag. I think it might be a reactor bug, but thought I'd ask here first before posting a bug.
For context: I have a cache Map<Key, Mono<Value>>. A client will request data - we check the cache and use what is essentially computeIfAbsent to place a Mono with .cache() into the cache if nothing has yet been cached for that key. The client then takes the Mono and does magic (not relevant here). Now, the catch is that the population of the cache may encounter transient errors, so we don't want to cache errors - the current request will error but the "next" client, when it subscribes, should trigger the entire pipeline to rerun.
Having read around, for example this closed issue, I settled on Mono#cache(ttlForValue, ttlForError, ttlForEmpty).
This is where things get interesting.
As I don't want to cache error (or empty, but ignore that) signals I found the following documentation promising
If the relevant TTL generator throws any
Exception, that exception will be propagated to the Subscriber that encountered the cache miss, but the cache will be immediately cleared, so further Subscribers might re-populate the cache in case the error was transient. In case the source was emitting an error, that error is dropped and added as a suppressed exception. In case the source was emitting a value, that value is dropped.
emphasis mine
So I tried the following (shamelessly cribbing the example in the linked GitHub issue)
public class TestBench {
public static void main(String[] args) throws Exception {
var sampleService = new SampleService();
var producer = Mono.fromSupplier(sampleService::call).cache(
__ -> Duration.ofHours(24),
//don't cache errors
e -> {throw Exceptions.propagate(e);},
//meh
() -> {throw new RuntimeException();});
try {
producer.block();
} catch (RuntimeException e) {
System.out.println("Caught exception : " + e);
}
sampleService.serverAvailable = true;
var result = producer.block();
System.out.println(result);
}
static final class SampleService {
volatile boolean serverAvailable = false;
String call() {
System.out.println("Calling service with availability: " + serverAvailable);
if (!serverAvailable) throw new RuntimeException("Error");
return "Success";
}
}
}
Output
09:12:23.991 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
Calling service with availability: false
09:12:24.034 [main] ERROR reactor.core.publisher.Operators - Operator called default onErrorDropped
java.lang.RuntimeException: Error
at uk.co.borismorris.testbench.TestBench$SampleService.call(TestBench.java:40)
at reactor.core.publisher.MonoSupplier.subscribe(MonoSupplier.java:56)
at reactor.core.publisher.MonoCacheTime.subscribe(MonoCacheTime.java:123)
at reactor.core.publisher.Mono.block(Mono.java:1474)
at uk.co.borismorris..boris.testbench.TestBench.main(TestBench.java:26)
Caught exception : reactor.core.Exceptions$BubblingException: java.lang.RuntimeException: Error
Exception in thread "main" java.lang.RuntimeException: Error
at uk.co.borismorris.testbench.TestBench$SampleService.call(TestBench.java:40)
at reactor.core.publisher.MonoSupplier.subscribe(MonoSupplier.java:56)
at reactor.core.publisher.MonoCacheTime.subscribe(MonoCacheTime.java:123)
at reactor.core.publisher.Mono.block(Mono.java:1474)
at uk.co.borismorris.testbench.TestBench.main(TestBench.java:26)
Suppressed: java.lang.Exception: #block terminated with an error
at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:93)
at reactor.core.publisher.Mono.block(Mono.java:1475)
at uk.co.borismorris.testbench.TestBench.main(TestBench.java:31)
Well, that didn't work - the error is cached and the second subscriber just sees the same error.
Looking at the code the cause is obvious
Duration ttl = null;
try {
ttl = main.ttlGenerator.apply(signal);
}
catch (Throwable generatorError) {
signalToPropagate = Signal.error(generatorError);
STATE.set(main, signalToPropagate); //HERE
if (signal.isOnError()) {
//noinspection ThrowableNotThrown
Exceptions.addSuppressed(generatorError, signal.getThrowable());
}
}
The STATE is set to the error signal, not cleared at all. But this isn't the whole story,
the reason for the code not clearing the cache is below this block
if (ttl != null) {
main.clock.schedule(main, ttl.toMillis(), TimeUnit.MILLISECONDS);
}
else {
//error during TTL generation, signal != updatedSignal, aka dropped
if (signal.isOnNext()) {
Operators.onNextDropped(signal.get(), currentContext());
}
else if (signal.isOnError()) {
Operators.onErrorDropped(signal.getThrowable(), currentContext());
}
//immediate cache clear
main.run();
}
In this case ttl == null because the generation of the ttl threw an Exception. The signal is an error so that branch is entered and Operators.onErrorDropped is called
public static void onErrorDropped(Throwable e, Context context) {
Consumer<? super Throwable> hook = context.getOrDefault(Hooks.KEY_ON_ERROR_DROPPED,null);
if (hook == null) {
hook = Hooks.onErrorDroppedHook;
}
if (hook == null) {
log.error("Operator called default onErrorDropped", e);
throw Exceptions.bubble(e);
}
hook.accept(e);
}
So here we can see that if there is no onError hook in the context and no default set then throw Exceptions.bubble(e) is called and the code in MonoCacheTime returns early, failing to call main.run(). Hence the error stays cached indefinitely as there is no TTL!
The following code fixes that problem
public class TestBench {
private static final Logger logger = LoggerFactory.getLogger(TestBench.class);
private static final Consumer<Throwable> onErrorDropped = e -> logger.error("Dropped", e);
static {
//add default hook
Hooks.onErrorDropped(onErrorDropped);
}
public static void main(String[] args) throws Exception {
var sampleService = new SampleService();
var producer = Mono.fromSupplier(sampleService::call).cache(
__ -> Duration.ofHours(24),
//don't cache errors
e -> {throw Exceptions.propagate(e);},
//meh
() -> {throw new RuntimeException();});
try {
producer.block();
} catch (RuntimeException e) {
System.out.println("Caught exception : " + e);
}
sampleService.serverAvailable = true;
var result = producer.block();
System.out.println(result);
}
static final class SampleService {
volatile boolean serverAvailable = false;
String call() {
System.out.println("Calling service with availability: " + serverAvailable);
if (!serverAvailable) throw new RuntimeException("Error");
return "Success";
}
}
}
But this adds a global Hook, which isn't ideal. The code hints at the ability to add per-pipeline hooks, but I cannot figure out how to do that. The following works, but is obviously a hack
.subscriberContext(ctx -> ctx.put("reactor.onErrorDropped.local", onErrorDropped))
Questions
- Is the above a bug, should the absence of a
onErrorDroppedhook cause errors to be cached indefinitely? - Is there a way to set the
onErrorDroppedhook in thesubscriberContextrather than globally?
Follow up
From the code; it seems that returning null from a TTL generator function is supported and has the same behaviour when the signal is immediately cleared. In the case where it isn't, the subscriber sees the original error rather than the error from the TTL generator and a suppressed error - which seems perhaps neater
public static void main(String[] args) throws Exception {
var sampleService = new SampleService();
var producer = Mono.fromSupplier(sampleService::call).cache(
__ -> Duration.ofHours(24),
//don't cache errors
e -> null,
//meh
() -> null);
try {
producer.block();
} catch (RuntimeException e) {
System.out.println("Caught exception : " + e);
}
sampleService.serverAvailable = true;
var result = producer.block();
System.out.println(result);
}
Is this behaviour supported? Should it be documented?