Unclear documentation about quiet time for RedisClient#shutdown

[LychakGalina]

Bug Report

Current Behavior

When calling redisClient#shutdown with quite period, channel writer state is marked to Closed immediately (io.lettuce.core.protocol.DefaultEndpoint#closeAsync) and when new request is submitted during quite period (it's still not expired) we get "io.lettuce.core.RedisException: Connection is closed"

Stack trace

io.lettuce.core.RedisException: Connection is closed
	at io.lettuce.core.protocol.DefaultEndpoint.validateWrite(DefaultEndpoint.java:195)
	at io.lettuce.core.protocol.DefaultEndpoint.write(DefaultEndpoint.java:137)
	at io.lettuce.core.protocol.CommandExpiryWriter.write(CommandExpiryWriter.java:112)
	at io.lettuce.core.RedisChannelHandler.dispatch(RedisChannelHandler.java:187)
	at io.lettuce.core.StatefulRedisConnectionImpl.dispatch(StatefulRedisConnectionImpl.java:152)
	at io.lettuce.core.RedisPublisher$RedisSubscription.dispatchCommand(RedisPublisher.java:385)
	at io.lettuce.core.RedisPublisher$CommandDispatch$1.dispatch(RedisPublisher.java:455)
	at io.lettuce.core.RedisPublisher$RedisSubscription.checkCommandDispatch(RedisPublisher.java:380)
	at io.lettuce.core.RedisPublisher$State$2.request(RedisPublisher.java:521)
	at io.lettuce.core.RedisPublisher$RedisSubscription.request(RedisPublisher.java:238)
	at reactor.core.publisher.MonoNext$NextSubscriber.request(MonoNext.java:102)
	at reactor.core.publisher.MonoNext$NextSubscriber.request(MonoNext.java:102)
	at reactor.core.publisher.StrictSubscriber.onSubscribe(StrictSubscriber.java:77)
	at reactor.core.publisher.MonoNext$NextSubscriber.onSubscribe(MonoNext.java:64)
	at reactor.core.publisher.MonoNext$NextSubscriber.onSubscribe(MonoNext.java:64)
	at io.lettuce.core.RedisPublisher$State$1.subscribe(RedisPublisher.java:500)
	at io.lettuce.core.RedisPublisher$RedisSubscription.subscribe(RedisPublisher.java:221)
	at io.lettuce.core.RedisPublisher.subscribe(RedisPublisher.java:127)
	at reactor.core.publisher.MonoFromPublisher.subscribe(MonoFromPublisher.java:55)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4105)
	at io.reactivex.internal.operators.observable.ObservableFromPublisher.subscribeActual(ObservableFromPublisher.java:31)
	at io.reactivex.Observable.subscribe(Observable.java:12090)
	at io.reactivex.internal.operators.observable.ObservableMap.subscribeActual(ObservableMap.java:32)
	at io.reactivex.Observable.subscribe(Observable.java:12090)
	at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
	at io.reactivex.Observable.subscribe(Observable.java:12090)
	at io.reactivex.internal.operators.observable.ObservableSingleSingle.subscribeActual(ObservableSingleSingle.java:35)
	at io.reactivex.Single.subscribe(Single.java:3438)
	at io.reactivex.internal.operators.single.SingleFlatMap$SingleFlatMapCallback.onSuccess(SingleFlatMap.java:84)
	at io.reactivex.internal.operators.single.SingleMap$MapSingleObserver.onSuccess(SingleMap.java:64)
	at io.reactivex.internal.operators.single.SingleDoOnSubscribe$DoOnSubscribeSingleObserver.onSuccess(SingleDoOnSubscribe.java:77)
	at io.reactivex.internal.operators.single.SingleJust.subscribeActual(SingleJust.java:30)
	at io.reactivex.Single.subscribe(Single.java:3438)
	at io.reactivex.internal.operators.single.SingleDoOnSubscribe.subscribeActual(SingleDoOnSubscribe.java:41)
	at io.reactivex.Single.subscribe(Single.java:3438)
	at io.reactivex.internal.operators.single.SingleMap.subscribeActual(SingleMap.java:34)
	at io.reactivex.Single.subscribe(Single.java:3438)
	at io.reactivex.internal.operators.single.SingleFlatMap.subscribeActual(SingleFlatMap.java:36)
	at io.reactivex.Single.subscribe(Single.java:3438)
	at io.reactivex.internal.operators.single.SingleToFlowable.subscribeActual(SingleToFlowable.java:37)
	at io.reactivex.Flowable.subscribe(Flowable.java:14479)
	at io.reactivex.Flowable.subscribe(Flowable.java:14429)
	at reactor.core.publisher.MonoFromPublisher.subscribe(MonoFromPublisher.java:55)
	at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55)
	at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
	at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1630)
	at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241)
	at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203)
	at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1630)
	at reactor.core.publisher.MonoIgnoreThen$ThenAcceptInner.onNext(MonoIgnoreThen.java:296)
	at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2186)
	at reactor.core.publisher.MonoIgnoreThen$ThenAcceptInner.onSubscribe(MonoIgnoreThen.java:285)
	at reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:191)
	at reactor.core.publisher.MonoFlatMap.subscribeOrReturn(MonoFlatMap.java:53)
	at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:48)
	at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:153)
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
	at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55)
	at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
	at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67)
	at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:76)
	at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:274)
	at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:851)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
	at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:173)
	at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2186)
	at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.request(MonoPeekTerminal.java:132)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1994)
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1868)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
	at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onSubscribe(MonoPeekTerminal.java:145)
	at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4105)
	at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:441)
	at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:211)
	at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:139)
	at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:63)
	at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55)
	at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
	at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55)
	at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
	at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55)
	at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4105)
	at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.drain(MonoIgnoreThen.java:172)
	at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:56)
	at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55)
	at reactor.netty.http.server.HttpServerHandle.onStateChange(HttpServerHandle.java:64)
	at reactor.netty.tcp.TcpServerBind$ChildObserver.onStateChange(TcpServerBind.java:226)
	at reactor.netty.http.server.HttpServerOperations.onInboundNext(HttpServerOperations.java:441)
	at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:89)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	at reactor.netty.http.server.HttpTrafficHandler.channelRead(HttpTrafficHandler.java:167)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:438)
	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:326)
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:300)
	at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1422)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:931)
	at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:792)
	at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:502)
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:407)
	at io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1050)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:834)

Input Code

Input Code

        redisClient.shutdown(5, 10, TimeUnit.SECONDS);

Expected behavior/code

According Netty documentation, requests submitted during the quite period should be processed:

If a task is submitted during the quiet period, it is guaranteed to be accepted and the quiet period will start over.

Btw, is it worth adding a description for quite period from the documentation, because current description of parameter looks not clear?

* @param quietPeriod the quiet period as described in the documentation

Environment

• Lettuce version(s): 5.2.1.RELEASE
• Redis version: 5.0.0

Possible Solution

Additional context

[mp911de]

Timeouts on AbstractRedisClient.shutdown(…) are a legacy from the time when a single EventLoopGroup as associated with the RedisClient instance. That is when the EventLoopGroup lifecycle is bound to the lifecycle of a RedisClient instance.

We should update our documentation to reflect this aspect. We're closing all connections when calling shutdown(…) to avoid lingering open channels and there's not really something we could do about it.

[LychakGalina]

Thanks for answer. But not sure that realize the problem why shutdown couldn't be delayed for quite period? and eventLoopProvider is still shutdown with a quiet period anyway.

But if so, shouldn't they be rather marked as deprecated and removed in the next release?

[mp911de]

What is your use case for delaying the shutdown and requiring a driver to set up a timer for the actual resource disposal?

Although typical applications control the shutdown of eventloops via ClientResources, we still allow creation of the client with a ClientResources object whose lifecycle is controlled by RedisClient. While that’s it the recommended approach, it’s not that bad that it would require deprecation and removal.

[LychakGalina]

we use redisClient and spring boot with embedded server that has graceful shutdown, so when sigterm is received it waits for all in-progress requests are finished, and we have situation when redisClient is destroyed before request is finished and some of them are failed with java.util.concurrent.RejectedExecutionException: event executor terminated.
And here we have two options - make dependency webServerFactory bean to redisClient and destroy it after, or make some delay within shutdown of redisClient (it doesn't guarantee that we get rid of such exceptions, but minimize them at least

we still allow creation of the client with a ClientResources object whose lifecycle is controlled by RedisClient.

it will not help, since clientResources is terminated already with a quite period, but channelWriter is not.

[mp911de]

Can you also raise the same issue in the Spring Boot issue tracker? It could make sense to define shutdown dependencies to ensure the frontend stops accepting web requests before other components get shut down.

[LychakGalina]

Raised spring-projects/spring-boot#19951,
but we don't have issues with accepting new requests since we deploy application to kubernetes and it removes endpoint from service right after sigterm is received, so the problem we have mostly about current requests that already started before sigterm.

[mp911de]

Thanks. Graceful removal and taking traffic away from a service is an entirely different discussion. That concern should be rather handled on an application level where the service gets disabled (i.e. excluded from the loadbalancing group) before the actual shutdown.

We're going to update the documentation as mentioned.

[mp911de]

Docs are updated now.