reactor/reactor-netty

Reactive Network Driver based on Netty and Reactor

nkorrapati

@nkorrapati

ok thanks

any estimate on that date may be in this month

Violeta Georgieva

@violetagg

it is scheduled for November 12 - https://github.com/spring-projects/spring-boot/milestone/178

nkorrapati

@nkorrapati

thanks

Rafael Ferreira

@rafasf

o/ everyone -- I've been looking for for a way to use lifo as the leasing strategy when using WebClient. Is there a programmatic way or is the system property the only way for now?

Violeta Georgieva

@violetagg

@rafasf you have to use ReactorResourceFactory https://docs.spring.io/spring-framework/docs/current/reference/html/web-reactive.html#webflux-client-builder-reactor-resources

22 replies

Rafael Ferreira

@rafasf

Oh, sorry.. I guess I can't put images in threads

Mark Rotteveel

@mrotteveel

I was wondering, if I'm creating a ReferenceCounted object in a decoder, and the object is emitted by the flux from NettyInbound.receiveObject(), at what point should I call retain() on it? Do I call retain() in the decoder, or do I call it after the object was emitted by the flux (or could that result in a race condition with the release() in FluxReceive.drainReceiver())? Do I even need to call retain on it?

2 replies

Yifu Diao

@yifudiao_twitter

We are seeting "reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response" in our production environment but not our staging environment. We are wondering if something is wrong with our infrastructure. From your experience, what kind of infra configurations can cause this problem?

14 replies

James Bodkin

@JBodkin-LH

Is it possible to override the default behaviour in the Netty Server to disable keep-alive? If so, how can I do this? Our application is running inside of a Kubernetes cluster and we've been seeing an uneven spread of requests against our running pods as persistent connections prevent Kubernetes services from being able to load balance

1 reply

Sam Lightfoot

@samlighty_twitter

Hey guys. We've been seeing a lot of ReadTimeoutExceptions using Spring WebFlux, and wondered if it made any sense for the Spring WebClient to use its own EventLoopGroup rather than share the same worker group as the Netty server? The WebClient appears to be starving the server from request processing.

3 replies

acerdas84

@acerdas84

I was googling around trying to understand what is Netty max connections, I mean how many connections can a single server can take at a given time? (like for tomcat which defaults to 8192 in spring-boot), but I have not found anything on that, do you know or can point me to the right place to check that??

Peter

@warmuuh

Hi. before netty 1.0.0, we used tcpServer::bootstrap to register a gauge for pending task queue of the eventloop

     factory.addServerCustomizers(httpServer -> httpServer.tcpConfiguration(
         tcpServer -> tcpServer.bootstrap(serverBootstrap -> {
               serverBootstrap.config().childGroup().forEach(eventExecutor -> {
                 if (SingleThreadEventExecutor.class.isAssignableFrom(eventExecutor.getClass())) {
                   SingleThreadEventExecutor singleThreadEventExecutor = (SingleThreadEventExecutor) eventExecutor;

                   Gauge.builder(SERVER_PENDING_TASK_METRIC, singleThreadEventExecutor::pendingTasks)
                       .description("Pending Tasks")
                       .tag(SERVER_THREAD_NAME, singleThreadEventExecutor.threadProperties().name())
                       .tag(SERVER_THREAD_STATE, singleThreadEventExecutor.threadProperties().state().name())
                       .register(registry);
                 }
               });
               return serverBootstrap;
             }
         )));

starting with netty 1.0.0, bootstrap is not available anymore. how can this be done now? thx for help

11 replies

Prathamesh Bongale

@prathambongale

Hi Team,

I am trying below code in Spring WebFlux but it runs 2 times. I am using Spring Boot 2.3.3.RELEASE and mongodb as database.

There are 3 repositories called: sessionRepo, userRepo, requestRepo

return sessionRepo
    .findBySid(request.getId())
    .log()
    .flatMap(sessionInfo -> userRepo.findById(sessionInfo.getUid())
            .log()
            .map(quoteDocument -> {
    // other update user logic
                userRepo.save(user).subscribe();
    return success;
            })
            .onErrorReturn().defaultIfEmpty())
    .switchIfEmpty(createNewUser(request))
    .onErrorReturn());

public Mono<Success> createNewUser(Request request){
return requestRepo.findById(request.getRequestInfo().getId())
.log()
.map(requestInfo -> {
// other create user logic
userRepo.save(user).subscribe();
return success;
})
.defaultIfEmpty()
.onErrorReturn();
} ```

I have noticed most of my code runs 2 times. But all the API’s where I must create new ID’s like session id and send it to Front end it always sends the second one creating duplicate in database.

Even switchIfEmpty runs 2 times.

3 replies

Pavel Kuchin

@pavelkuchin

Hi here! We are using reactor-netty webclient and webclient handgs up at arbitrary moments. What is interesting we are using multiple instances of webclient in out app and it affects all the instances. Usually it happens only on one instance and at arbitrary moments. We can't connect it to high load or any other metrics in our system except cpu consumption and netty active connections metric. Cpu consumption starts growing gradually meantime active.connections growing as well up to max connections. Right after it reaches max value pending.connections metric starts growing. Based on metrics it seems like active connections freeze for some reason and all requests are going to pending queue until they canceled by timeout. Elevated CPU consumption is caused by reactor-http-epoll threads. I've enabled debug logging for netty but so far can't find anything interesting there.

Any advise?

Pavel Kuchin

@pavelkuchin

What is interesting It started last Saturday (last deployment we had was at Wednesday before) and now happens pretty regularly (at least once a day). We are using reactor-netty-0.9.14 at the moment.

Pavel Kuchin

@pavelkuchin

It seems I found moment in logs when zombie connection appears, looks the same way as when current channel closes an new one opens, except instead of 1 there are 2 open now, the zombie channels are growing overtime exhausting connection pool:

11/29/20
7:38:19.513 PM    
2020-11-30 00:38:19,513 +0000 [reactor-http-epoll-1] [DEBUG] (r.n.r.PooledConnectionProvider:?) [8bbde608-86d7-40e2-90ee-6b84797e9243#jkto] - [id: 0xe31e9eb0, L:/masked:masked - R:masked/masked:masked] Channel connected, now 2 active connections and 0 inactive connections
host = masked/c03060e213b3 source = masked.log sourcetype = log4j
11/29/20
7:38:19.513 PM    
2020-11-30 00:38:19,513 +0000 [reactor-http-epoll-1] [DEBUG] (r.n.r.PooledConnectionProvider:?) [8bbde608-86d7-40e2-90ee-6b84797e9243#jkto] - [id: 0xe31e9eb0, L:/masked:masked - R:masked/masked:masked] Registering pool release on close event for channel
host = masked/c03060e213b3 source = masked.log sourcetype = log4j
11/29/20
7:38:19.511 PM    
2020-11-30 00:38:19,511 +0000 [reactor-http-epoll-1] [DEBUG] (reactor.netty.tcp.SslProvider:?) [44e53d93-1496-4b4c-bf3c-c528d6ac4a16#un4m] - [id: 0xe31e9eb0] SSL enabled using engine SSLEngineImpl and SNI masked:masked
host = masked/c03060e213b3 source = masked.log sourcetype = log4j
11/29/20
7:37:58.133 PM    
2020-11-30 00:37:58,133 +0000 [reactor-http-epoll-3] [DEBUG] (r.n.r.PooledConnectionProvider:?) [bd61425b-58c0-4482-beb8-6970913b46cb#hzsb] - [id: 0x675d970b, L:/masked:masked ! R:masked/masked:masked] onStateChange(PooledConnection{channel=[id: 0x675d970b, L:/masked:masked ! R:masked/masked:masked]}, [disconnecting])
host = masked/c03060e213b3 source = masked.log sourcetype = log4j
11/29/20
7:37:27.964 PM    
2020-11-30 00:37:27,964 +0000 [reactor-http-epoll-3] [DEBUG] (r.n.r.PooledConnectionProvider:?) [bd61425b-58c0-4482-beb8-6970913b46cb#hzsb] - [id: 0x675d970b, L:/masked:masked - R:masked/masked:masked] Channel cleaned, now 0 active connections and 1 inactive connections
host = masked/c03060e213b3 source = masked.log sourcetype = log4j
11/29/20
7:37:27.963 PM    
2020-11-30 00:37:27,963 +0000 [reactor-http-epoll-3] [DEBUG] (r.n.r.PooledConnectionProvider:?) [bd61425b-58c0-4482-beb8-6970913b46cb#hzsb] - [id: 0x675d970b, L:/masked:masked - R:masked/masked:masked] Releasing channel

3 replies

mbrkic4

@mbrkic4

Hi, I've noticed that client callbacks doOnRedirect and doOnRequest (for redirected request when redirect following is enabled) sometimes run on different threads, so the order in which they are called (and related end time) is not deterministic. Is that by design? I'm registering all durations and statuses of an upstream request in Spring Cloud Gateway, but now I have synchronization issues (same data structure that stores durations and statuses is modified concurrently) and also can't be sure about the order (e.g. 200 might be registered before 307 even if they happened in inverse order). If reactor netty doesn't wait after onRedirect is finished, before proceeding with redirected request, how should I handle such issues?

4 replies

Ismail Marmoush

@IsmailMarmoush

Hi,
Is there a way I could synchronize incoming requests of the same request param value ?
Example: assuming /cmd/{id} any incoming requests /cmd/1 would be running on same thread and one after another in the order they came in with.

devansh-dalal

@devansh-dalal

I am using netty HttpClient with proxy configuration, its working fine from my mac with config

HttpClient.create()
            .wiretap(true)
            .secure(
                ssl -> {
                  ClassLoader classLoader = App.class.getClassLoader();
                  var context = SslContextBuilder.forClient().trustManager(getSslCert());
                  ssl.sslContext(context);
                })
            .proxy(
                p ->
                    p.type(Proxy.HTTP)
                        .host("3.215.204.48"/*or "https://tntzruoqxoy.sandbox.verygoodproxy.com" */)
                        .port(8080)
           );

The proxy host can be just IP address or Url. Both works fine.
But when I try to set host=url in docker I get this exception

Caused by: org.springframework.web.reactive.function.client.WebClientRequestException: nested exception is java.nio.channels.UnresolvedAddressException
    at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:137) ~[spring-webflux-5.3.1.jar!/:5.3.1]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    |_ checkpoint ⇢ Request to GET https://httpbin.org/get [DefaultWebClient]
Stack trace:
        at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:137) ~[spring-webflux-5.3.1.jar!/:5.3.1]
        at reactor.core.publisher.MonoErrorSupplied.subscribe(MonoErrorSupplied.java:70) ~[reactor-core-3.4.0.jar!/:3.4.0]
        at reactor.core.publisher.Mono.subscribe(Mono.java:3987) ~[reactor-core-3.4.0.jar!/:3.4.0]
        at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103) ~[reactor-core-3.4.0.jar!/:3.4.0]
        at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:221) ~[reactor-core-3.4.0.jar!/:3.4.0]

Things work fine when I set proxy host as IP address in docker or I don't use the proxy altogether.
So, somehow proxy host url is not getting resolved in docker. Has anyone seen this issue before?
docker: docker container run --interactive --tty --rm openjdk:11-jre-slim bash

5 replies

Vonorims

@Vonorims

Hey, I'm pretty new with reactive programming and I met the problem.

reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response

I know that it's a famous problem and I tried to solve it by myself with github/stackoverflow issues and topics, but it didn't help. So, what I have.
A non-reactive token REST-service (POST request -> response which contains token) and main reactive service to collect some statistics.

My logs with error:

2020-12-02 16:19:25,777 INFO  [scheduling-1] com.mir.apigoogleanalytics.service.DataUpdateService: Scheduled view id collecting has been started.
2020-12-02 16:19:26,782 INFO  [scheduling-1] com.mir.apigoogleanalytics.service.GaViewIdCollectorService: Collecting viewId for account: some_account@gmail.com
2020-12-02 16:19:27,503 DEBUG [reactor-http-nio-1] reactor.util.Loggers$Slf4JLogger: [id: 0xef935ed9, L:/ 10.1.10.200:5888 - R:access-token-example.com/172.190.220.28:443] Handler is being applied: {uri=https://access-token-example.com/oauth/actual_token, method=POST}
2020-12-02 16:19:27,808 DEBUG [reactor-http-nio-1] reactor.util.Loggers$Slf4JLogger: [id: 0xef935ed9, L:/ 10.1.10.200:5888 - R:access-token-example.com/172.190.220.28:443] Received response (auto-read:false) : [Server=nginx/1.15.3, Date=Wed, 02 Dec 2020 13:19:28 GMT, Content-Type=application/json, Transfer-Encoding=chunked, Connection=keep-alive, Vary=Accept-Encoding]
2020-12-02 16:19:27,833 DEBUG [reactor-http-nio-1] reactor.util.Loggers$Slf4JLogger: [id: 0xef935ed9, L:/ 10.1.10.200:5888 - R:access-token-example.com/172.190.220.28:443] Received last HTTP packet
2020-12-02 16:19:27,919 DEBUG [reactor-http-nio-1] reactor.util.Loggers$Slf4JLogger: [id: 0xcf973ffd, L:/192.168.1.58:5889 - R:www.googleapis.com/64.233.162.95:443] Handler is being applied: {uri=https://www.googleapis.com/analytics/v3/management/accounts/~all/webproperties/~all/profiles/~all/goals, method=GET}
2020-12-02 16:19:29,334 DEBUG [reactor-http-nio-1] reactor.util.Loggers$Slf4JLogger: [id: 0xcf973ffd, L:/192.168.1.58:5889 - R:www.googleapis.com/64.233.162.95:443] Received response (auto-read:false) : [ETag="VWwXKed-YkV-a1QGNAfzEn91uZm216yTisCT7qwkUg8/nWQDQveNldnouLyOvLbTF7nqnSY", Content-Type=application/json; charset=UTF-8, Vary=Origin, Vary=X-Origin, Vary=Referer, Date=Wed, 02 Dec 2020 13:19:29 GMT, Server=ESF, Cache-Control=private, X-XSS-Protection=0, X-Frame-Options=SAMEORIGIN, X-Content-Type-Options=nosniff, Alt-Svc=h3-29=":443"; ma=2592000,h3-T051=":443"; ma=2592000,h3-Q050=":443"; ma=2592000,h3-Q046=":443"; ma=2592000,h3-Q043=":443"; ma=2592000,quic=":443"; ma=2592000; v="46,43", Transfer-Encoding=chunked]
2020-12-02 16:19:29,520 DEBUG [reactor-http-nio-1] reactor.util.Loggers$Slf4JLogger: [id: 0xcf973ffd, L:/192.168.1.58:5889 - R:www.googleapis.com/64.233.162.95:443] Received last HTTP packet
2020-12-02 16:25:45,647 INFO  [reactor-http-nio-1] com.mir.apigoogleanalytics.service.GaViewIdCollectorService: Collecting viewId for account: another_account@gmail.com
2020-12-02 16:25:45,654 DEBUG [reactor-http-nio-1] reactor.util.Loggers$Slf4JLogger: [id: 0xef935ed9, L:/ 10.1.10.200:5888 - R:access-token-example.com/172.190.220.28:443] Handler is being applied: {uri=https://access-token-example.com/oauth/actual_token, method=POST}
2020-12-02 16:25:45,670 WARN  [reactor-http-nio-1] reactor.util.Loggers$Slf4JLogger: [id: 0xef935ed9, L:/ 10.1.10.200:5888 ! R:access-token-example.com/172.190.220.28:443] The connection observed an error
reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response
2020-12-02 16:25:45,675 WARN  [reactor-http-nio-1] com.mir.apigoogleanalytics.config.reactor.RetryConfig: Retrying: 0; 0; {}
reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    |_ checkpoint ⇢ Request to POST null [DefaultWebClient]

6 replies

Anil Gursel

@anilgursel

@violetagg Is there an example/doc of using Reactor Netty + OpenSSL with session based resumption?

3 replies

Diwan

@diwanoli

@violetagg I am currently using JettyReactorConnector instead of ReactorClientHttpConnector . I am getting below error

reactor.core.Exceptions$BubblingException: java.lang.IllegalStateException: No thread-bound request found: Are you referring to request attributes outside of an actual web request, or processing a request outside of the originally receiving thread? If you are actually operating within a web request and still receive this message, your code is probably running outside of DispatcherServlet: In this case, use RequestContextListener or RequestContextFilter to expose the current request.

2 replies

Sam Xiao

@samxiao_gitlab

Hello @violetagg

Do you have any good tips/recommendation to dive deeper into these exceptions?

I posted in stackoverflow

2020-12-11 00:29:12.498 ERROR 11918 --- [ctor-http-nio-2] r.n.channel.ChannelOperationsHandler     : [id: 0xebbe5eff, L:0.0.0.0/0.0.0.0:55189] Error was received while reading the incoming data. The connection will be closed.

io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
        at io.netty.util.internal.ReferenceCountUpdater.toLiveRealRefCnt(ReferenceCountUpdater.java:74)
        at io.netty.util.internal.ReferenceCountUpdater.release(ReferenceCountUpdater.java:138)
        at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:100)
        at io.netty.handler.codec.http.DefaultHttpContent.release(DefaultHttpContent.java:92)
        at io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:88)
        at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:340)
        at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:358)
        at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:647)
        at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:96)
        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.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
        at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
        at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
        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.ssl.SslHandler.unwrap(SslHandler.java:1526)

27 replies

Johannes Conen

@TheConen

Can someone help me getting the human readable logging mentioned in reactor/reactor-netty#233 and reactor/reactor-netty#1270 to work? I'm a bit lost there and the documentation (https://projectreactor.io/docs/netty/snapshot/reference/index.html) doesn't mention anything about it.

I'm currently using Spring WebFlux and with logging.level.reactor.netty.http.client.HttpClient = DEBUG in application.propertiesand wiretap enabled (see code snippet below) I get the hex dump logging. Does anyone know what I need to change in order to get the human readable logging?

private WebClient webClient = WebClient.builder()
            .clientConnector(new ReactorClientHttpConnector(
                    HttpClient.create().wiretap(true)
            ))
            .baseUrl(someUri)
            .build();

8 replies

ctlove0523

@ctlove0523

Is it possible to close connection when server receive SYN from client?

pippoflow

@pippoflow

@violetagg If i need to add a custom decoder before reactor bridge, how do I go about doing this? In the decoder I would already decode ByteBuf to my own object, and from my understanding bytebufExtractor will throw exception when ChannelOperationsHandler calls channelRead?

10 replies

pippoflow

@pippoflow

Is it possible to replace the subscriber of inbound? For different types of data flowing in the channel, I want different subscribers to handle it.

iron2414

@iron2414

Hello, i'm trying to connect to a Mysql database with Spring webflux, using r2dbc and jasync. The application starts fine, however if i make a http call then i get the following netty error:

2020-12-28 16:51:44.722  WARN 10432 --- [ctor-http-nio-3] io.netty.channel.ChannelInitializer      : Failed to initialize a channel. Closing: [id: 0xa8528cbb, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:61020]

I'm trying to connect to a XAMPP mysql server, fully default. Database and table exists.

Here is the sample, just clone it, run it and make a http call to http://localhost:8080/user/test

https://github.com/iron2414/webflux-mysql

6 replies

Bing Ran/冉兵

@branaway

Is there a sample Restful server based on Reactor-netty kicking around? I'd like to see how different methods are properly handled. Plus I'd like to see how cache is controlled using the API. Thanks!

Bing Ran/冉兵

@branaway

For example, I could not figure out how to get the request content as a string or InputStream

Violeta Georgieva

@violetagg

@branaway Do you want pure Reactor Netty or Spring WebFlux? With Spring WebFlux you have a variaty of encoders/decoders.

harrisrahman

@harrisrahman

Hello, I have a very simple Spring Webflux application on reactor-netty. The RestController method takes no argument , and returns a Mono<String> - very simple application.

I am running performance test using gatling and performance seems to be very poor. I see a lot of

j.i.IOException: Premature close
j.n.SocketException: Connection reset by peer
A 99 percentile of 11 seconds for such a plain service says definitely I am doing something wrong in configuring netty. Please suggest if there is any netty configuration that I am missing ?

harrisrahman

@harrisrahman

Here are netty startup logs

2021-01-02 22:34:37.861 DEBUG 29069 --- [ main] r.netty.resources.DefaultLoopEpoll : Default Epoll support : false
2021-01-02 22:34:37.863 DEBUG 29069 --- [ main] r.netty.resources.DefaultLoopKQueue : Default KQueue support : false
2021-01-02 22:34:37.871 DEBUG 29069 --- [ main] i.n.channel.MultithreadEventLoopGroup : -Dio.netty.eventLoopThreads: 24
2021-01-02 22:34:37.899 DEBUG 29069 --- [ main] i.n.u.internal.InternalThreadLocalMap : -Dio.netty.threadLocalMap.stringBuilder.initialSize: 1024
2021-01-02 22:34:37.899 DEBUG 29069 --- [ main] i.n.u.internal.InternalThreadLocalMap : -Dio.netty.threadLocalMap.stringBuilder.maxSize: 4096
2021-01-02 22:34:37.908 DEBUG 29069 --- [ main] io.netty.channel.nio.NioEventLoop : -Dio.netty.noKeySetOptimization: false
2021-01-02 22:34:37.908 DEBUG 29069 --- [ main] io.netty.channel.nio.NioEventLoop : -Dio.netty.selectorAutoRebuildThreshold: 512
2021-01-02 22:34:37.919 DEBUG 29069 --- [ main] i.netty.util.internal.PlatformDependent : org.jctools-core.MpscChunkedArrayQueue: available
2021-01-02 22:34:37.987 DEBUG 29069 --- [ main] io.netty.util.ResourceLeakDetector : -Dio.netty.leakDetection.level: simple
2021-01-02 22:34:37.987 DEBUG 29069 --- [ main] io.netty.util.ResourceLeakDetector : -Dio.netty.leakDetection.targetRecords: 4
2021-01-02 22:34:38.124 DEBUG 29069 --- [ main] io.netty.channel.DefaultChannelId : -Dio.netty.processId: 29069 (auto-detected)
2021-01-02 22:34:38.130 DEBUG 29069 --- [ main] io.netty.channel.DefaultChannelId : -Dio.netty.machineId: 38:f9:d3:ff:fe:22:64:e8 (auto-detected)
2021-01-02 22:34:38.166 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.numHeapArenas: 24
2021-01-02 22:34:38.166 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.numDirectArenas: 24
2021-01-02 22:34:38.166 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.pageSize: 8192
2021-01-02 22:34:38.166 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.maxOrder: 11
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.chunkSize: 16777216
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.tinyCacheSize: 512
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.smallCacheSize: 256
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.normalCacheSize: 64
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.maxCachedBufferCapacity: 32768
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.cacheTrimInterval: 8192
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.cacheTrimIntervalMillis: 0
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.useCacheForAllThreads: true
2021-01-02 22:34:38.167 DEBUG 29069 --- [ main] io.netty.buffer.PooledByteBufAllocator : -Dio.netty.allocator.maxCachedByteBuffersPerChunk: 1023
2021-01-02 22:34:38.175 DEBUG 29069 --- [ main] io.netty.buffer.ByteBufUtil : -Dio.netty.allocator.type: pooled
2021-01-02 22:34:38.175 DEBUG 29069 --- [ main] io.netty.buffer.ByteBufUtil : -Dio.netty.threadLocalDirectBufferSize: 0
2021-01-02 22:34:38.176 DEBUG 29069 --- [ main] io.netty.buffer.ByteBufUtil : -Dio.netty.maxThreadLocalCharBufferSize: 16384

harrisrahman

@harrisrahman

Controller code

@RestController
class PosOrderController {

    @RequestMapping(value = ["/v1/pos/order-detail"], method = [RequestMethod.GET])
    fun getDetail(@RequestParam(required = true) language: String): Mono<String> {
        val response = Mono.just("Hello").subscribeOn(Schedulers.boundedElastic())
        response.publishOn(Schedulers.boundedElastic()).subscribe()
        return response
    }
}

harrisrahman

@harrisrahman

Also see lots of java.lang.NoClassDefFoundError: Could not initialize class org.springframework.web.util.HtmlUtils
SpringBoot : 2.4.1
Reactor-netty : 1.0.2

harrisrahman

@harrisrahman

Here is the log for a single connection - I am thinking the slowness is due to open/close connection for every request, could that be the case ?

2021-01-03 03:09:58.187 DEBUG 35142 --- [ctor-http-nio-6] r.n.http.server.HttpServerOperations : [id: 0xf8717c47, L:/127.0.0.1:9090 - R:/127.0.0.1:53761] New http connection, requesting read
2021-01-03 03:09:58.187 DEBUG 35142 --- [ctor-http-nio-6] reactor.netty.transport.TransportConfig : [id: 0xf8717c47, L:/127.0.0.1:9090 - R:/127.0.0.1:53761] Initialized pipeline DefaultChannelPipeline{(reactor.left.httpCodec = io.netty.handler.codec.http.HttpServerCodec), (reactor.left.httpTrafficHandler = reactor.netty.http.server.HttpTrafficHandler), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}
2021-01-03 03:09:58.188 DEBUG 35142 --- [ctor-http-nio-6] r.n.http.server.HttpServerOperations : [id: 0xf8717c47, L:/127.0.0.1:9090 - R:/127.0.0.1:53761] Increasing pending responses, now 1
2021-01-03 03:09:58.188 DEBUG 35142 --- [ctor-http-nio-6] reactor.netty.http.server.HttpServer : [id: 0xf8717c47, L:/127.0.0.1:9090 - R:/127.0.0.1:53761] Handler is being applied: org.springframework.http.server.reactive.ReactorHttpHandlerAdapter@6388d854
2021-01-03 03:09:58.188 DEBUG 35142 --- [ctor-http-nio-6] o.s.w.s.adapter.HttpWebHandlerAdapter : [f8717c47-197] HTTP GET "/v1/pos/order-detail?language=en&svc_version=1.0"
2021-01-03 03:09:58.188 DEBUG 35142 --- [ctor-http-nio-6] s.w.r.r.m.a.RequestMappingHandlerMapping : [f8717c47-197] Mapped to com.loblaw.operations.flapi.comsumers.sweda.PosOrderController#getDetail(String)
2021-01-03 03:09:58.189 DEBUG 35142 --- [ctor-http-nio-6] o.s.w.r.r.m.a.ResponseBodyResultHandler : [f8717c47-197] Using 'text/html' given [text/html, application/xhtml+xml, application/xml;q=0.9, /;q=0.8] and supported [text/plain;charset=UTF-8, text/event-stream, text/plain;charset=UTF-8, /]
2021-01-03 03:09:58.190 DEBUG 35142 --- [ctor-http-nio-6] o.s.w.r.r.m.a.ResponseBodyResultHandler : [f8717c47-197] 0..1 [java.lang.String]
2021-01-03 03:09:58.190 DEBUG 35142 --- [undedElastic-25] o.s.core.codec.CharSequenceEncoder : [f8717c47-197] Writing "Hello"
2021-01-03 03:09:58.191 DEBUG 35142 --- [ctor-http-nio-6] r.n.http.server.HttpServerOperations : [id: 0xf8717c47, L:/127.0.0.1:9090 - R:/127.0.0.1:53761] Decreasing pending responses, now 0
2021-01-03 03:09:58.192 DEBUG 35142 --- [ctor-http-nio-6] o.s.w.s.adapter.HttpWebHandlerAdapter : [f8717c47-197] Completed 200 OK
2021-01-03 03:09:58.192 DEBUG 35142 --- [ctor-http-nio-6] r.n.http.server.HttpServerOperations : [id: 0xf8717c47, L:/127.0.0.1:9090 - R:/127.0.0.1:53761] Last HTTP response frame
2021-01-03 03:09:58.192 DEBUG 35142 --- [ctor-http-nio-6] r.n.http.server.HttpServerOperations : [id: 0xf8717c47, L:/127.0.0.1:9090 - R:/127.0.0.1:53761] Last HTTP packet was sent, terminating the channel

Violeta Georgieva

@violetagg

@harrisrahman Here you can find my experiments with Spring and Gatling (I don’t know why you use additional thread for sending a String but you may have more complex scenario)
https://github.com/violetagg/web-benchmarks/blob/3869db0dd18a66cc0280be1bed45553ed073a2d9/webflux-benchmark/src/main/java/io/spring/sample/webfluxbenchmark/MyRestController.java#L36
https://github.com/violetagg/web-benchmarks/blob/3869db0dd18a66cc0280be1bed45553ed073a2d9/webfluxfn-benchmark/src/main/java/io/spring/sample/webfluxfnbenchmark/RouterFunctionConfig.java#L60
https://github.com/violetagg/web-benchmarks/blob/3869db0dd18a66cc0280be1bed45553ed073a2d9/containers/client/gatling/simulations/webbenchmark/TextPlain.scala
https://youtu.be/Y2wMPG-htpE?list=ULdJYHeRDxD5g&t=1364

1 reply

harrisrahman

@harrisrahman

@violetagg , thank you, will take a look.

harrisrahman

@harrisrahman

idikshit

@idikshit

I was trying to figure out a way to silently discard an incoming HTTP Request without sending back any response. Is there any way I can achieve this in Netty Server?. Thanks in advance.

13 replies

Anders Clausen

@AndersClausen

Hi @violetagg . We have noticed that a request of size 1 is being initiated when a stream of Flux is requested, followed by a request of 127. Is there a specific reason for this behaviour? We notice it when using WebFlux but reckons it's a Netty piece of logic? Many thanks

10 replies

Kushagra Thapar

@kushagraThapar

@violetagg - after upgrading reactor-netty from 0.9.13 to 1.0.1, I am noticing connection pool is getting created multiple times, even though I am only creating the http client once in my application. Earlier I was using fixed API, now I am passing the maxConnectionPoolSize in the builder to create HttpClient. My use-case is to re-use the fixed number of connections throughout my application, which was successfully working on version 0.9.x. But on version 1.x.x - I doubt if that is happening, as why would there be need to create client pool again and again ?
I am seeing this in my debug logs multiple times for a single http client.

2021-01-12 12:51:07,838       [parallel-1] DEBUG reactor.netty.resources.PooledConnectionProvider - Creating a new [reactor-netty-connection-pool] client pool [PoolFactory{evictionInterval=PT0S, leasingStrategy=fifo, maxConnections=1000, maxIdleTime=60000, maxLifeTime=-1, metricsEnabled=false, pendingAcquireMaxCount=-1, pendingAcquireTimeout=45000}] for

10 replies

prashantsc

@prashantsc

Hi,
I have some unexpected behavior with reactor-netty in WebClient context I have set up the WebClient with ConnectionPrivider as follows : and my connection-timeout is 15000ms.

LCE [qtp1928983164-55] DEBUG r.n.r.PooledConnectionProvider - Creating a new [webclient-pool] client pool [PoolFactory{evictionInterval=PT0S, leasingStrategy=fifo, maxConnections=128, maxIdleTime=5000, maxLifeTime=5000, metricsEnabled=true, pendingAcquireMaxCount=2048, pendingAcquireTimeout=30000}] for [my-service:8080]

It does honor maxIdleTime set to 5000ms idle time, I see the connections even after 5s and they get cleaned up after 30s (no mater what value maxIdleTime is set to).
Please help debugging this issue.
Versions in use:

'io.projectreactor.addons', name: 'reactor-extra', version: '3.3.4.RELEASE'
'io.projectreactor.netty', name: 'reactor-netty', version: '0.9.15.RELEASE'
'io.projectreactor', name: 'reactor-core', version: '3.3.12.RELEASE'
springBootVersion = 2.2.1.RELEASE

3 replies

Anders Clausen

@AndersClausen

@violetagg Regarding my post on Friday, did you see that I provided some code snippets as per your request? I was just wondering if you had an answer to my question. Cheers

Violeta Georgieva

@violetagg

@AndersClausen I will reply in the thread above

Mateusz Morawski

@MorasOne_twitter

Hey, I've an question about Flux.

I need to receive asynchronously some object, lets' name it "Object A", then I need to receive another object asynchronously, let's name it "Object B".
When I finish processing of object B, I need to do something on "Object A" - always, especially when something goes wrong while receiving/processing "Object B". I wonder, how to acheive this without netsing flatMaps.
Bellow my working snippet:

    serviceA.tryReceive()
            .flatMapMany(objectA ->
                    serviceB.tryReceive()
                            .flatMap(objectB -> {
                                //do some async stuff
                            })
                            .doFinally(signalType -> {
                                serviceA.doSmth(objectA);
                            }))
            .subscribe();

Where communities thrive

People

Repo info

Activity