reactor-netty: SSLException on request cancellation

In our SDK we are seeing javax.net.ssl.SSLException: SSLEngine closed already exceptions. We are using fixed channel pool - ConnectionProvider with 1000 as pool size.

Expected Behavior

On cancel operation, channel should be closed / disposed quietly since the operation has been cancelled. Whether that channel is reusable or not can be a point of discussion.

Actual Behavior

Combination of pagination + takeUntil operator causes SSLException. In high load, the error keeps happening, and thus causing Out of Memory issue.

Steps to Reproduce

I was able to reproduce the issue on an isolated project - which is exactly same we have reactor-netty http client in Azure CosmosDB Java SDK.

Here is the link to the project I have created : https://github.com/kushagraThapar/reactor-netty-ssl-error

Run the MainApplication.java in IDE for couple of minutes and the error starts happening.

For testing purpose, I have kept the connection pool size to 5.

Your Environment

macOS / linux machines reactor-netty: 0.9.7.RELEASE reactor-core: 3.3.5.RELEASE

Here is the complete stack trace:

18 Jun 2020 16:04:31,770       [reactor-http-nio-12] WARN reactor.netty.http.client.HttpClientConnect - [id: 0xbb8e3529, L:/192.168.0.10:54545 - R:rel.ink/104.24.122.41:443] The connection observed an error
javax.net.ssl.SSLException: SSLEngine closed already
	at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:848)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoIgnoreThen] :
	reactor.core.publisher.Mono.thenEmpty
	reactor.netty.ReactorNetty$OutboundThen.<init>(ReactorNetty.java:553)
Error has been observed at the following site(s):
	|_  Mono.thenEmpty ⇢ at reactor.netty.ReactorNetty$OutboundThen.<init>(ReactorNetty.java:553)
	|_                 ⇢ at reactor.netty.ReactorNetty$OutboundThen.then(ReactorNetty.java:598)
	|_ Mono.fromDirect ⇢ at reactor.netty.http.client.HttpClientConnect$HttpIOHandlerObserver.lambda$onStateChange$0(HttpClientConnect.java:437)
	|_      Mono.defer ⇢ at reactor.netty.http.client.HttpClientConnect$HttpIOHandlerObserver.onStateChange(HttpClientConnect.java:437)
Stack trace:
		at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:848)
		at io.netty.handler.ssl.SslHandler.wrapAndFlush(SslHandler.java:811)
		at io.netty.handler.ssl.SslHandler.flush(SslHandler.java:792)
		at io.netty.handler.ssl.SslHandler.flush(SslHandler.java:1931)
		at io.netty.handler.ssl.SslHandler.closeOutboundAndChannel(SslHandler.java:1899)
		at io.netty.handler.ssl.SslHandler.close(SslHandler.java:743)
		at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:622)
		at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:606)
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.close(CombinedChannelDuplexHandler.java:505)
		at io.netty.channel.ChannelOutboundHandlerAdapter.close(ChannelOutboundHandlerAdapter.java:77)
		at io.netty.channel.CombinedChannelDuplexHandler.close(CombinedChannelDuplexHandler.java:316)
		at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:622)
		at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:606)
		at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:472)
		at io.netty.channel.DefaultChannelPipeline.close(DefaultChannelPipeline.java:957)
		at io.netty.channel.AbstractChannel.close(AbstractChannel.java:232)
		at reactor.netty.http.client.HttpClientOperations.onInboundCancel(HttpClientOperations.java:258)
		at reactor.netty.channel.FluxReceive.unsubscribeReceiver(FluxReceive.java:436)
		at reactor.netty.channel.FluxReceive.lambda$new$0(FluxReceive.java:80)
		at reactor.netty.channel.FluxReceive.cancelReceiver(FluxReceive.java:148)
		at reactor.netty.channel.FluxReceive.cancel(FluxReceive.java:90)
		at reactor.netty.channel.ChannelOperations.dispose(ChannelOperations.java:160)
		at reactor.core.publisher.MonoCreate$DefaultMonoSink.onCancel(MonoCreate.java:222)
		at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect$TcpClientSubscriber.onNext(HttpClientConnect.java:344)
		at reactor.netty.http.client.HttpClientConnect$MonoHttpConnect$TcpClientSubscriber.onNext(HttpClientConnect.java:329)
		at reactor.core.publisher.MonoCreate$DefaultMonoSink.success(MonoCreate.java:156)
		at reactor.netty.resources.PooledConnectionProvider$DisposableAcquire.run(PooledConnectionProvider.java:604)
		at reactor.netty.resources.PooledConnectionProvider$DisposableAcquire.onNext(PooledConnectionProvider.java:485)
		at reactor.netty.resources.PooledConnectionProvider$DisposableAcquire.onNext(PooledConnectionProvider.java:436)
		at reactor.netty.internal.shaded.reactor.pool.AbstractPool$Borrower.deliver(AbstractPool.java:388)
		at reactor.netty.internal.shaded.reactor.pool.SimplePool.lambda$drainLoop$13(SimplePool.java:244)
		at reactor.core.scheduler.ImmediateScheduler.schedule(ImmediateScheduler.java:47)
		at reactor.netty.internal.shaded.reactor.pool.SimplePool.drainLoop(SimplePool.java:244)
		at reactor.netty.internal.shaded.reactor.pool.SimplePool.drain(SimplePool.java:172)
		at reactor.netty.internal.shaded.reactor.pool.SimplePool.maybeRecycleAndDrain(SimplePool.java:159)
		at reactor.netty.internal.shaded.reactor.pool.SimplePool$QueuePoolRecyclerInner.onComplete(SimplePool.java:401)
		at reactor.core.publisher.Operators.complete(Operators.java:135)
		at reactor.core.publisher.MonoEmpty.subscribe(MonoEmpty.java:45)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4218)
		at reactor.netty.internal.shaded.reactor.pool.SimplePool$QueuePoolRecyclerMono.subscribe(SimplePool.java:452)
		at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4218)
		at reactor.core.publisher.Mono.subscribeWith(Mono.java:4329)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4189)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4125)
		at reactor.netty.resources.PooledConnectionProvider$PooledConnection.onStateChange(PooledConnectionProvider.java:401)
		at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:426)
		at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:607)
		at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:96)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
		at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
		at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
		at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1518)
		at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1267)
		at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1314)
		at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:501)
		at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:440)
		at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
		at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
		at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
		at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
		at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)
		at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:714)
		at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650)
		at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576)
		at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
		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)