SendUtils onDiscard handler class cast exception

[Svenskunganka]

I'm using RSocket Java with Spring Webflux and RSocket JS in the front-end, using REQUEST_STREAM with an infinite Kafka source. The returned Flux receives quite a lot of small messages and does quite a bit of filtering and buffering. The relevant code in SendUtils does an unchecked cast of each discarded object to a ReferenceCounted which seems to throw a class cast exception that is caught and ignored.
The problem is that due to the sheer amount of objects being discarded (due to filtering and discarded buffers on disconnect), this uses up a substantial amount of CPU resources looking at the profiler.

The returned Fluxes are scheduled on Schedulers.boundedElastic(), and for some reason entire threads are started up just for this exception:

Expected Behavior

Not use up so much CPU resources

Actual Behavior

Uses a lot of CPU resources

Steps to Reproduce

Possible Solution

Not exactly sure, but perhaps an instanceof check:

if (data instanceof ReferenceCounted) ((ReferenceCounted) data).release();

Your Environment

• RSocket version(s) used: rsocket-core:1.1.3, rsocket-transport-netty:1.1.3
• Other relevant libraries versions (eg. netty, ...): Spring Webflux 5.3.23
• Platform (eg. JVM version (javar -version) or Node version (node --version)): 17
• OS and version (eg uname -a): Happens on all tested platforms (Debian, Mac)

[OlegDokuka]

@Svenskunganka! Sounds relevant!

Feel free to contribute if you see the direct solution, otherwise, I will investigate it!

Best,
Oleh

[Svenskunganka]

Hi Oleh!

If my possible solution is acceptable I can definitely contribute it! But if you perhaps know of another way this could be solved without needing to perform an instanceof check for every discarded object that might be a better route to explore.
From my limited experience the only thing I can think of is using e.g Mono.using()/Flux.using() when creating a Payload.

input.flatMap(bytes -> Mono.using(
    () -> ByteBufPayload.create(bytes),
    data -> ...,
    Payload::release
));

Kind regards,
Tom

[luczus]

Hi Oleh, @Svenskunganka!

I've created PR #1079 with a fix.
Please let me know what you think about these changes.

Best
Lukasz

[tukez]

I noticed the same problem and in my case at least the offending instance is of type org.springframework.core.io.buffer.NettyDataBuffer which holds a ByteBuf. Should the inner ByteBuf be released? That is not possible (other than reflection) in rsocket-java though?

[kmccarp]

@OlegDokuka are there any plans to integrate the changes made by @luczus? This is heavily polluting our logs.