IllegalReferenceCountException with reactor-netty 0.6.1

[jakobklein]

I use reactor's TcpClient to connect to a Server.
While exchanging messages I get very often a IllegalReferenceCountException (see below) from netty.
I didn't have this problem with [core: 3.0.1.RELEASE, netty: 0.5.2.RELEASE].
It started after upgrading to core: [3.0.5.RELEASE, netty: 0.6.1.RELEASE]

Is it a bug or am I doing something wrong?
Maybe someone can give me a hint from the attached logs...

IllegalReferenceCountException.txt

[jakobklein]

SMaldini found a workaround for my case:
To receive a string with the TcpClient I had to add .retain():
in.receive().retain().asString().subscribe(incomingMsg ->

It's still a workaround and I think code/api/doc need some improvement because it's not obvious why one would need to use .retain() here.

Here's the gitter-conversation (23.2.2017) that lead to the workaround:

Jakob Klein @jakobklein 15:12
While using TcpClient to exchange messages with a server I often get a IllegalReferenceCountException from netty. Just wanted to ask if this is a known issue with reactor-netty 0.6.1 or if someone can give me a hint what might be going wrong. I filed it with log and stacktrace: #51

Stephane Maldini @smaldini 15:15
@jakobklein do you use code like echo inbound to outbound ?
e.g. send(r.receive())
or alternatively do you consume ByteBuf yourself directly ?
like receive().buffer().doOnNext(Consumer<list>)
the ref counting is supposed to be more accurate but that means something we tolerated before might not work
we work with Netty ref counting around ByteBuf , meaning that we pre allocate bytes and we receive them from netty with +1, we forward to onNext in receive(), if direct we release (-1) if buffered due to missing backpressure we dont -1
that means that if you have receive().collectList() for instance, we will do -1 while you wanted to use these buffers still
for this reason we offer receive().retain() operator

Stephane Maldini @smaldini 15:20
that will auto + 1 but you have to -1 yourself or by writing the packets (also -1 in this)
its a difference wrt rxnetty which by default retain all and avoid this kind of issue
but you might only detect later memory leak
its hard enough to debug memory leak so we are defensive by default

Jakob Klein @jakobklein 15:26
I do not consume ByteBuf directly. I consume strings.
About echoing: I do something similar, but don't know if this is relevant:
in.receive().asString().subscribe(incomingMsg -> {telegramService.consume(...

Stephane Maldini @smaldini 15:28
if you immediately put them as String you escape any management of byte buf
but why would that fail tho
saaaad. we ran a bunch of integration before 0.6.1

Jakob Klein @jakobklein 15:28
where TelegramService does something like: out.sendString(Mono.just(tgString)).subscribe(
don't know if this is still what you call echoing.

Stephane Maldini @smaldini 15:29
its something similar yeah but since you manipulate String
you copied the values over in memory
we dont care about the underlying ByteBuf

Stephane Maldini @smaldini 15:34
thats very strange
is it only TCP ? or HttpClient ?
can you ry adding retain() before asString() ?
and if you have extra context about the conditions this occur (or if it looks regular on any kind of flow or a specific one)
good thing that 0.6.2 will add metrics/serviceability -_-

Jakob Klein @jakobklein 16:16
It's only TCP.
I had a hard time reducing my test to a minimum...
but retain() before asString() seems to do the trick. I'm running some more tests.

Jakob Klein @jakobklein 16:41
yes, looks good. the problem seems to be solved.
Thanks for your help. I will add our conversation to the github-issue because even if the workaround works, it is not very obvious, why I would need retain() with .asString().