RedisReactiveHealthIndicator makes blocking call on error

[ayudovin]

See #16705

---

Caused by: java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-3
	at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:77)
	at reactor.core.publisher.Mono.block(Mono.java:1494)
	at org.springframework.data.redis.connection.ReactiveRedisConnection.close(ReactiveRedisConnection.java:60)
	at org.springframework.boot.actuate.redis.RedisReactiveHealthIndicator.lambda$doHealthCheck$1(RedisReactiveHealthIndicator.java:50)
	at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.runFinally(FluxDoFinally.java:156)
	... 203 common frames omitted

While running a Spring Cloud Gateway application with Redis as session store, I observed the stack trace above. The cause is a blocking call in: https://github.com/spring-projects/spring-boot/blob/master/spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/redis/RedisReactiveHealthIndicator.java#L49

There is a Mono<Void> closeLater(); https://github.com/spring-projects/spring-data-redis/blob/master/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java#L68 which could be called instead.

[wilkinsona]

Thanks for the PR, @ayudovin. As mentioned by @mp911de, we also need to change how connections are obtained. Of the two suggestions that Mark's made, I think the use of the parallel scheduler is the one to go for. Would you like to update your PR to that effect?

[mp911de]

I filed DATAREDIS-977 to expose non-blocking connect methods. Right now, Mono.fromSupplier().subscribeOn(Schedulers.parallel()) or similar should be a good compromise until we can provide Mono<ReactiveRedisConnection> methods.

[ayudovin]

@wilkinsona, I would like to update PR

[AndreasKl]

@ayudovin The Mono returned via closeLater is unhandled, as it is lazy the close call will never happen.

  @Test
  public void notWorking() {
    Mono.fromRunnable(() -> System.out.println("not emitted"));
  }

  @Test
  public void working() {
    Mono<Void> never = Mono.fromRunnable(() -> System.out.println("working"));
    never.block();
  }

https://github.com/spring-projects/spring-data-redis/blob/master/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java#L239

[philwebb]

Thanks once again @ayudovin!