Skip to content

report leak when client readtimeout. #1408

Closed
Closed
report leak when client readtimeout.#1408
Labels
for/springframeworkThis belongs to the Spring Framework project
@yinjianfei

Description

@yinjianfei
yinjianfei
opened on Dec 4, 2020

Expected behavior

No leak

Actual behavior

LEAK: ByteBuf.release() was not called before it's garbage-collected. See http://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
#1:
	io.netty.buffer.AdvancedLeakAwareByteBuf.writeBytes(AdvancedLeakAwareByteBuf.java:616)
	io.netty.buffer.ByteBufOutputStream.write(ByteBufOutputStream.java:68)
	com.fasterxml.jackson.core.json.UTF8JsonGenerator._flushBuffer(UTF8JsonGenerator.java:2085)
	com.fasterxml.jackson.core.json.UTF8JsonGenerator.flush(UTF8JsonGenerator.java:1097)
	com.fasterxml.jackson.databind.ObjectWriter.writeValue(ObjectWriter.java:915)
	org.springframework.http.codec.json.AbstractJackson2Encoder.encodeValue(AbstractJackson2Encoder.java:174)
	org.springframework.http.codec.json.AbstractJackson2Encoder.lambda$encode$0(AbstractJackson2Encoder.java:122)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107)
	reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2070)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1878)
	reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1752)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
	reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
	reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
	reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
	reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
	reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
	reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
	reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2070)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	reactor.core.publisher.MonoFlatMap$FlatMapInner.onSubscribe(MonoFlatMap.java:230)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
	reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
	reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
	reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:144)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoSingle$SingleSubscriber.onComplete(MonoSingle.java:171)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.checkTerminated(FluxFlatMap.java:794)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:560)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:540)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.onComplete(FluxFlatMap.java:426)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.checkTerminated(FluxFlatMap.java:794)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:560)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:540)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.onComplete(FluxFlatMap.java:426)
	reactor.core.publisher.DrainUtils.postCompleteDrain(DrainUtils.java:131)
	reactor.core.publisher.DrainUtils.postComplete(DrainUtils.java:186)
	reactor.core.publisher.FluxMapSignal$FluxMapSignalSubscriber.onComplete(FluxMapSignal.java:213)
	reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136)
	reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:252)
	reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:252)
	reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136)
	reactor.netty.channel.FluxReceive.terminateReceiver(FluxReceive.java:372)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:196)
	reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:337)
	reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:333)
	reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:382)
	reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:522)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:141)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:345)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:337)
	io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:438)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:323)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:297)
	io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:345)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:337)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:345)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:796)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
	java.lang.Thread.run(Thread.java:748)
Created at:
	io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:334)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:173)
	io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:107)
	org.springframework.core.io.buffer.NettyDataBufferFactory.allocateBuffer(NettyDataBufferFactory.java:65)
	org.springframework.core.io.buffer.NettyDataBufferFactory.allocateBuffer(NettyDataBufferFactory.java:39)
	org.springframework.http.codec.json.AbstractJackson2Encoder.encodeValue(AbstractJackson2Encoder.java:168)
	org.springframework.http.codec.json.AbstractJackson2Encoder.lambda$encode$0(AbstractJackson2Encoder.java:122)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107)
	reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2070)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:1878)
	reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:1752)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
	reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
	reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
	reactor.core.publisher.MonoSwitchIfEmpty.subscribe(MonoSwitchIfEmpty.java:44)
	reactor.core.publisher.MonoFlatMap.subscribe(MonoFlatMap.java:60)
	reactor.core.publisher.MonoOnErrorResume.subscribe(MonoOnErrorResume.java:44)
	reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
	reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2070)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	reactor.core.publisher.MonoFlatMap$FlatMapInner.onSubscribe(MonoFlatMap.java:230)
	reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
	reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
	reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
	reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:144)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241)
	reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
	reactor.core.publisher.MonoSingle$SingleSubscriber.onComplete(MonoSingle.java:171)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.checkTerminated(FluxFlatMap.java:794)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:560)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:540)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.onComplete(FluxFlatMap.java:426)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.checkTerminated(FluxFlatMap.java:794)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:560)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:540)
	reactor.core.publisher.FluxFlatMap$FlatMapMain.onComplete(FluxFlatMap.java:426)
	reactor.core.publisher.DrainUtils.postCompleteDrain(DrainUtils.java:131)
	reactor.core.publisher.DrainUtils.postComplete(DrainUtils.java:186)
	reactor.core.publisher.FluxMapSignal$FluxMapSignalSubscriber.onComplete(FluxMapSignal.java:213)
	reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136)
	reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:252)
	reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:252)
	reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136)
	reactor.netty.channel.FluxReceive.terminateReceiver(FluxReceive.java:372)
	reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:196)
	reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:337)
	reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:333)
	reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:382)
	reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:522)
	reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:141)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:345)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:337)
	io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:438)
	io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:323)
	io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:297)
	io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:345)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:337)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:359)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:345)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
	io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:796)
	io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
	io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
	io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
	java.lang.Thread.run(Thread.java:748)

Steps to reproduce

call /leak/test serval times with set client side readtimeout to 1ms.

Reproducable code

@Component
public class Handler {

    private WebClient webClient;

    public Handler(WebClient.Builder webClientBuilder) {
        String url = UriComponentsBuilder.newInstance()
                .scheme("http")
                .host("127.0.0.1")
                .build()
                .toUriString();
        webClient = webClientBuilder
                .baseUrl(url)
                .clientConnector(new ReactorClientHttpConnector(HttpClient.create()))
                .build();
    }
    public Mono<ServerResponse> handle(ServerRequest serverRequest) {
        return query(serverRequest.path()).flatMap(jsonNode -> ServerResponse.ok().syncBody(jsonNode));
    }
    private Mono<JsonNode> query(String path) {
        return webClient.get()
                .uri(uriBuilder -> uriBuilder
                        .port(8080)
                        .path(path)
                        .build())
                .retrieve()
                .bodyToMono(JsonNode.class)
                .flatMap(Mono::just);
    }
}


public class RouterConfig {

    @Bean
    @Order(1)
    public RouterFunction<ServerResponse> route(Handler handler) {
        return RouterFunctions.route()
                .GET("/leak/test", handler::handle)
                .build();
    }

}

logs with setting -Dio.netty.leakDetectionLevel=paranoid, reactor.netty=debug at: https://github.com/yinjianfei/webflux-leak-test/blob/main/console.log
code repo at: https://github.com/yinjianfei/webflux-leak-test

java --version

openjdk version "1.8.0_275"

springboot version

2.1.4.RELEASE

Metadata

Metadata

Assignees

No one assigned

    Labels

    for/springframeworkThis belongs to the Spring Framework project

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions