Reactor CacheMono onCacheMissResume called when cache is present

Question

I want to use cachemono and created a test to evaluate. So I created 2 mono and concatenated. What I notice is that oncachemissresume is called for both mono, but on the second mono, cache is not written. Is there anything wrong with my test? I want to use oncachemissresume to populate the cache and the current test results does not help me.

AtomicReference<Context> storeRef = new AtomicReference<>(Context.empty());

        String key = "myid";
        Mono<Integer> cachedMonoFirst = CacheMono
                .lookup(k -> Mono.justOrEmpty(storeRef.get().<Integer>getOrEmpty(k))
                                .map(integer -> {
                                    log.error("first :: cache lookup result {}", integer);
                                    return Signal.next(integer);
                                }),
                        key)
                .onCacheMissResume(() -> {
                    log.info("first :: cache missed ");
                    return Mono.just(123);})
                .andWriteWith((k, sig) -> {
                    log.info("first :: cache write");
                    return Mono.fromRunnable(() ->
                        storeRef.updateAndGet(ctx -> ctx.put(k, sig.get())));});

        Mono<Integer> cachedMonoSecond = CacheMono
                .lookup(k -> Mono.justOrEmpty(storeRef.get().<Integer>getOrEmpty(k))
                                .map(integer -> {
                                    log.error("second :: cache lookup result {}", integer);
                                    return Signal.next(integer);
                                }),
                        key)
                .onCacheMissResume(() -> {
                    log.error("second :: cache missed");
                    return Mono.just(456);})
                .andWriteWith((k, sig) -> {
                    log.info("second :: cache write");
                    return Mono.fromRunnable(() ->
                        storeRef.updateAndGet(ctx -> ctx.put(k, sig.get())));});

        Flux<Integer> cacheFlux = cachedMonoFirst.concatWith(cachedMonoSecond);
        StepVerifier
                .create(cacheFlux)
                .consumeNextWith(data -> {
                    assertThat(storeRef.get().<Integer>getOrEmpty(key)).get().isEqualTo(data);
                    log.info(" first :: from cache {} {}", data, storeRef.get().<Integer>getOrEmpty(key));
                })
                .consumeNextWith(data -> {
                    assertThat(storeRef.get().<Integer>getOrEmpty(key)).get().isEqualTo(data);
                    log.info(" second :: from cache {} {}", data, storeRef.get().<Integer>getOrEmpty(key));
                })
                .verifyComplete();        

The logs are as follows

CacheTest - first :: cache missed 
CacheTest - first :: cache write
CacheTest -  first :: from cache 123 Optional[123]
CacheTest - second :: cache missed
CacheTest - second :: cache lookup result 123
CacheTest -  second :: from cache 123 Optional[123]

why does second :: cache missed called before second :: cache lookup result 123 ?

Answer 1

wrap onCacheMissResume with Mono.defer like shown below

.onCacheMissResume(Mono.defer(() -> {
                    log.error("second :: cache missed");
                    return Mono.just(456);}))