IllegalReferenceCountException happens when the operation was canceled

[violetagg]

This is a follow up to #22384

The following exception is observed

2019-03-14 08:09:37.211 ERROR 13233 --- [r-http-kqueue-7] reactor.core.publisher.Operators         : Operator called default onErrorDropped

io.netty.util.IllegalReferenceCountException: refCnt: 0
    at io.netty.buffer.AbstractByteBuf.ensureAccessible(AbstractByteBuf.java:1441)
    at io.netty.buffer.AbstractByteBuf.checkIndex(AbstractByteBuf.java:1373)
    at io.netty.buffer.PooledHeapByteBuf.nioBuffer(PooledHeapByteBuf.java:298)
    at io.netty.buffer.AbstractByteBuf.nioBuffer(AbstractByteBuf.java:1224)
    at io.netty.buffer.WrappedByteBuf.nioBuffer(WrappedByteBuf.java:919)
    at io.netty.buffer.AdvancedLeakAwareByteBuf.nioBuffer(AdvancedLeakAwareByteBuf.java:713)
    at org.springframework.core.io.buffer.NettyDataBuffer.asByteBuffer(NettyDataBuffer.java:266)
    at org.springframework.core.codec.StringDecoder.decodeDataBuffer(StringDecoder.java:207)
    at org.springframework.core.codec.StringDecoder.decodeDataBuffer(StringDecoder.java:59)
    at org.springframework.core.codec.AbstractDataBufferDecoder.lambda$decodeToMono$1(AbstractDataBufferDecoder.java:68)
    at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107)
    at reactor.core.publisher.FluxContextStart$ContextStartSubscriber.onNext(FluxContextStart.java:103)
    at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:287)
    at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:331)
    at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1505)
    at reactor.core.publisher.MonoCollectList$MonoBufferAllSubscriber.onComplete(MonoCollectList.java:123)
    at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:252)
    at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:252)
    at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136)
    at reactor.netty.channel.FluxReceive.terminateReceiver(FluxReceive.java:372)

The cause is the following:
https://github.com/spring-projects/spring-framework/blob/master/spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.java#L67

DataBufferUtils.join(InputStream) declares doOnDiscard, when such event happens the buffer will be recycled then when the StringDecoder tries to do the actual decoding the exception above will be observed

https://github.com/spring-projects/spring-framework/blob/master/spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java#L207

[rstoyanchev]

Aside from wrapping the error in something with a more helpful message, I just don't see what eles we can do. StringDecoder has to return something after all, an an error signal is the only thing that makes sense to propagate downstream. Such errors have to be expected, e.g. via onErrorResume, if a cancellation signal could come independently while reading. What do you think @violetagg, anything I'm missing?

[spring-projects-issues]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[spring-projects-issues]

Closing due to lack of requested feedback. If you would like us to look at this issue, please provide the requested information and we will re-open the issue.

[rstoyanchev]

Experimenting with something like this:

class LeakTests extends AbstractDataBufferAllocatingTests {

    @ParameterizedDataBufferAllocatingTest
    void byteCountsAndPositions(String displayName, DataBufferFactory bufferFactory) {
        super.bufferFactory = bufferFactory;

        try {
            Flux.range(1, 10)
                    .doOnNext(i -> {
                        if (i == 8) {
                            throw new IllegalStateException("boo");
                        }
                    })
                    .map(i -> bufferFactory.allocateBuffer())
                    .collectList()
                    .doOnDiscard(PooledDataBuffer.class, dataBuffer -> {
                        DataBufferUtils.release(dataBuffer);
                        dataBuffer.release();
                    })
                    .block();
        }
        catch (Exception ex) {
            // Ignore
        }
    }
}

I see that doOnDiscard does continue to iterate over the remaining buffers even after an IllegalReferenceCountException. However with reactor.core at WARN level you see:

14:44:54.836 [Test worker] WARN  r.c.p.Operators - Error while discarding element from a Collection, continuing with next element
io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
	at io.netty.util.internal.ReferenceCountUpdater.toLiveRealRefCnt(ReferenceCountUpdater.java:74) ~[netty-common-4.1.50.Final.jar:4.1.50.Final]
	at io.netty.util.internal.ReferenceCountUpdater.release(ReferenceCountUpdater.java:138) ~[netty-common-4.1.50.Final.jar:4.1.50.Final]
	at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:100) ~[netty-buffer-4.1.50.Final.jar:4.1.50.Final]
	at io.netty.buffer.WrappedByteBuf.release(WrappedByteBuf.java:1037) ~[netty-buffer-4.1.50.Final.jar:4.1.50.Final]
	at io.netty.buffer.SimpleLeakAwareByteBuf.release(SimpleLeakAwareByteBuf.java:102) ~[netty-buffer-4.1.50.Final.jar:4.1.50.Final]
	at io.netty.buffer.AdvancedLeakAwareByteBuf.release(AdvancedLeakAwareByteBuf.java:941) ~[netty-buffer-4.1.50.Final.jar:4.1.50.Final]
	at org.springframework.core.io.buffer.NettyDataBuffer.release(NettyDataBuffer.java:320) ~[main/:?]
	at org.springframework.core.io.buffer.LeakTests.lambda$byteCountsAndPositions$2(LeakTests.java:45) ~[test/:?]

So in the very least we can avoid the unnecessary stack trace and print a shorter message.