Fragmentation has been set to 15MB:
@Bean
public RSocketRequester getRSocketRequester(RSocketRequester.Builder builder, Flux<List<LoadbalanceTarget>> targetFlux) {
return builder
.rsocketConnector(rSocketConnector -> rSocketConnector
.fragment(15_728_640) //15MB
.reconnect(Retry.fixedDelay(2, Duration.ofSeconds(2))))
.setupMetadata(new UsernamePasswordMetadata(username, password), this.mimeType)
.rsocketStrategies(b -> b.encoder(new SimpleAuthenticationEncoder()))
.dataMimeType(MediaType.APPLICATION_CBOR)
.transports(targetFlux, new RoundRobinLoadbalanceStrategy());
}
But still, the following is thrown when the response is larger than 16MB:
Caused by: io.rsocket.exceptions.CanceledException: The payload is too big to be send as a single frame with a max frame length 16777215. Consider enabling fragmentation.
Version spring-boot-starter-rsocket:2.7.2
Full logs:
Caused by: io.rsocket.exceptions.CanceledException: The payload is too big to be send as a single frame with a max frame length 16777215. Consider enabling fragmentation.
at io.rsocket.exceptions.Exceptions.from(Exceptions.java:80)
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
*________________ ? at io.rsocket.core.RSocketRequester.requestResponse(RSocketRequester.java:125)
*________________ ? at io.rsocket.loadbalance.PooledRSocket.requestResponse(PooledRSocket.java:177)
|_ ? at io.rsocket.loadbalance.LoadbalanceRSocketClient.lambda$requestResponse$1(LoadbalanceRSocketClient.java:56)
*____Mono.flatMap ? at io.rsocket.loadbalance.LoadbalanceRSocketClient.requestResponse(LoadbalanceRSocketClient.java:56)
|_ Mono.map ? at org.springframework.messaging.rsocket.DefaultRSocketRequester$DefaultRequestSpec.retrieveMono(DefaultRSocketRequester.java:289)
|_ Mono.map ? at org.springframework.messaging.rsocket.DefaultRSocketRequester$DefaultRequestSpec.retrieveMono(DefaultRSocketRequester.java:290)
|_ ? at org.springframework.messaging.rsocket.DefaultRSocketRequester$DefaultRequestSpec.retrieveMono(DefaultRSocketRequester.java:272)
|_ ? at com.upstreamsystems.profiling.rsocket.client.profiles.ProfilesRSocketClient.searchProfiles(ProfilesRSocketClient.java:45)
|_ ? at com.upstreamsystems.profiling.apigateway.client.profiles.ProfilesAdapter.lambda$searchProfiles$7(ProfilesAdapter.java:86)
*____Mono.flatMap ? at com.upstreamsystems.profiling.apigateway.client.profiles.ProfilesAdapter.searchProfiles(ProfilesAdapter.java:86)
|_ Mono.retryWhen ? at com.upstreamsystems.profiling.apigateway.client.profiles.ProfilesAdapter.searchProfiles(ProfilesAdapter.java:87)
|_ Mono.handle ? at com.upstreamsystems.profiling.apigateway.client.profiles.ProfilesAdapter.searchProfiles(ProfilesAdapter.java:90)
|_ Mono.cast ? at com.upstreamsystems.profiling.apigateway.client.profiles.ProfilesAdapter.searchProfiles(ProfilesAdapter.java:91)
|_ ? at com.upstreamsystems.profiling.apigateway.service.ProfileIntegrationService.lambda$searchProfiles$19(ProfileIntegrationService.java:111)
*__Flux.concatMap ? at reactor.util.retry.RetryBackoffSpec.generateCompanion(RetryBackoffSpec.java:542)
*____Mono.flatMap ? at com.upstreamsystems.profiling.apigateway.service.ProfileIntegrationService.searchProfiles(ProfileIntegrationService.java:111)
Original Stack Trace:
at io.rsocket.exceptions.Exceptions.from(Exceptions.java:80)
at io.rsocket.core.RSocketRequester.handleFrame(RSocketRequester.java:261)
at io.rsocket.core.RSocketRequester.handleIncomingFrames(RSocketRequester.java:211)
at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160)
at io.rsocket.core.ClientServerInputMultiplexer$InternalDuplexConnection.onNext(ClientServerInputMultiplexer.java:248)
at io.rsocket.core.ClientServerInputMultiplexer.onNext(ClientServerInputMultiplexer.java:129)
at io.rsocket.core.ClientServerInputMultiplexer.onNext(ClientServerInputMultiplexer.java:48)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129)
at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:364)
at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:404)
at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93)
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.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)
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.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:800)
at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:487)
at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:385)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
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:833)
Comment From: bclozel
Is the fragmentation enabled on the server? There's a dedicated configuration property for that, spring.rsocket.server.fragment-size.
Comment From: Leuteris
Thank you Brian, it works. It seems that my previous configuration enables fragmentation on the client (request size).
Comment From: wilkinsona
Thanks for letting us know, @Leuteris.