Memory Leak in reactor-netty when using Spring WebClient with blocked flux. Issue: Databuffer is not released.

[KrishnakanthYachareni]

My service is responsible to consume the events from Kafka and will make a HTTP class to external server (REST API).

1. One Kafka event holds the list of 50 users
2. Once Kafka event is consumed by the service, it should make 50 HTTP parallel calls to external REST server and kafka thread should wait until all 50 user requests receive the response from the server.
3. Note: My application is not fully reactive, so I’m applying back-pressure by holding the Kafka thread until all 50 parallel HTTP calls are finished in one Kafka event.

Expected Behavior

There shouldn't be memory leaks when making multiple HTTP calls to external service.

Actual Behavior

When I did the performance testing, heap memory is not cleared. GC is not able to collect due to reactor DataBuff is not relasing the data.

Steps to Reproduce

I can't provide complete code to reproduce it but here's the main configuration.
WebClient Configuration

@Bean
    public WebClient webClient() {
        ConnectionProvider connectionProvider = getConnectionProvider();

        return WebClient.builder()
                .clientConnector(new ReactorClientHttpConnector(HttpClient.create(connectionProvider)
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) ofSeconds(5).toMillis()) // Connection timeout.
                        .responseTimeout(ofSeconds(5))
                        .keepAlive(true)
                        .protocol(HttpProtocol.H2)))
                .baseUrl("/base-url")
                .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .build();
    }

    private ConnectionProvider getConnectionProvider() {
        return ConnectionProvider.builder("fcmConn")
                .maxConnections(500)
                .maxIdleTime(ofSeconds(5)) 
                .maxLifeTime(ofSeconds(5))
                .pendingAcquireTimeout(ofSeconds(5))
                .evictInBackground(ofSeconds(5))
                .build();
    }

WebClient usage

private final ExecutorService executor = Executors.newCachedThreadPool();

 public Flux<FcmResponseWrapper> sendPush(Flux<Pair<FirebaseRequest, UserDeviceInfo>> messages, String appName) {
        try {
            String accessToken = fcmAuthenticationService.getAccessToken(appName);
            String appURI = fcmAuthenticationService.getAppURI(appName);

            return messages.flatMap(msgPair -> webClient.post()
                            .uri(appURI)
                            .header(HttpHeaders.AUTHORIZATION, accessToken)
                            .body(Mono.just(msgPair.getFirst()), FirebaseRequest.class)
                            .exchangeToFlux(clientResponse -> this.handleResponse(clientResponse, msgPair.getSecond())))
                    .subscribeOn(Schedulers.fromExecutor(executor));

        } catch (Exception exception) {
            // Logging
        }
    }
    
    private Flux<FcmResponseWrapper> handleResponse(ClientResponse response, UserDeviceInfo deviceInfo) {
        return response.bodyToMono(String.class)
                .<FcmResponseWrapper>handle((body, sink) -> {
                    try {
                        if (response.statusCode().is2xxSuccessful()) { // Extract the success response
                            SendResponse sendResponse = objectMapper.readValue(body, SendResponse.class);
                            sink.next(new FcmResponseWrapper(deviceInfo, sendResponse));
                        } else {
                            FirebaseErrorResponse errorResponse = objectMapper.readValue(body, FirebaseErrorResponse.class); // Extract the fcm error response
                            SendResponse sendResponse = SendResponse.getErrorResponse(errorResponse);
                            sink.next(new FcmResponseWrapper(deviceInfo, sendResponse));
                        }
                    } catch (Exception ex) {
                       // logging
                    }
                })
                .flux();
    }

How I'm calling the above method is:

Flux<FcmResponseWrapper> response = fcmRestClient.sendPush(fluxRequests, appName, eventIdentifiers);

List<Tuple2<Long, FcmResponseWrapper>> listOfSendResponses = response.elapsed()
                  .collectList()
                  .block(); // Thread blocking.

I tried to get the debug logs of LeakDetection but I couldn't find them in the logs

{"@timestamp":"2024-04-18T19:24:31.062-05:00","@version":"1","message":"[f64d0420] Created a new pooled channel, now: 0 active connections, 0 inactive connections and 0 pending acquire requests.","logger_name":"reactor.netty.resources.PooledConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.063-05:00","@version":"1","message":"[f64d0420] SSL enabled using engine io.netty.handler.ssl.JdkAlpnSslEngine@968d456 and SNI fcm.googleapis.com/<unresolved>:443","logger_name":"reactor.netty.tcp.SslProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.064-05:00","@version":"1","message":"[f64d0420] Initialized pipeline DefaultChannelPipeline{(reactor.left.sslHandler = io.netty.handler.ssl.SslHandler), (reactor.left.sslReader = reactor.netty.tcp.SslProvider$SslReadHandler), (reactor.left.h2Flush = io.netty.handler.flush.FlushConsolidationHandler), (reactor.left.httpCodec = io.netty.handler.codec.http2.Http2FrameCodec), (reactor.left.h2MultiplexHandler = io.netty.handler.codec.http2.Http2MultiplexHandler), (reactor.left.httpTrafficHandler = reactor.netty.http.client.HttpTrafficHandler), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}","logger_name":"reactor.netty.transport.TransportConfig","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.161-05:00","@version":"1","message":"[f64d0420] Connecting to [fcm.googleapis.com/216.239.36.57:443].","logger_name":"reactor.netty.transport.TransportConnector","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.225-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Registering pool release on close event for channel","logger_name":"reactor.netty.resources.DefaultPooledConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.225-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Channel connected, now: 1 active connections, 0 inactive connections and 0 pending acquire requests.","logger_name":"reactor.netty.resources.PooledConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.336-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] onStateChange(PooledConnection{channel=[id: 0xf64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443]}, [connected])","logger_name":"reactor.netty.resources.DefaultPooledConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.399-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] onStateChange(PooledConnection{channel=[id: 0xf64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443]}, [configured])","logger_name":"reactor.netty.resources.DefaultPooledConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.399-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Channel activated","logger_name":"reactor.netty.http.client.Http2Pool","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.399-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Channel deactivated","logger_name":"reactor.netty.http.client.Http2Pool","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.399-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443](H2 - -1) New HTTP/2 stream","logger_name":"reactor.netty.http.client.HttpClientOperations","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.399-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443](H2 - -1) Initialized HTTP/2 stream pipeline AbstractHttp2StreamChannel$3{(reactor.left.h2ToHttp11Codec = io.netty.handler.codec.http2.Http2StreamFrameToHttpObjectCodec), (reactor.left.httpTrafficHandler = reactor.netty.http.client.Http2StreamBridgeClientHandler), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}","logger_name":"reactor.netty.http.client.HttpClientConfig","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.399-05:00","@version":"1","message":"[f64d0420/1-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Handler is being applied: {uri=https://fcm.googleapis.com/v1/projects/wmt-sumo-pnp-poc/messages:send, method=POST}","logger_name":"reactor.netty.http.client.HttpClientConnect","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.405-05:00","@version":"1","message":"[f64d0420/1-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Stream opened, now: 1 active streams and 100 max active streams.","logger_name":"reactor.netty.http.client.Http2ConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.406-05:00","@version":"1","message":"[f64d0420/1-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Added encoder [reactor.left.responseTimeoutHandler] at the beginning of the user pipeline, full pipeline: [reactor.left.h2ToHttp11Codec, reactor.left.httpTrafficHandler, reactor.left.responseTimeoutHandler, reactor.right.reactiveBridge, DefaultChannelPipeline$TailContext#0]","logger_name":"reactor.netty.ReactorNetty","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.521-05:00","@version":"1","message":"[f64d0420/1-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Received response (auto-read:false) : RESPONSE(decodeResult: success, version: HTTP/1.1)\nHTTP/1.1 200 OK\ncontent-type: <filtered>\nvary: <filtered>\nvary: <filtered>\nvary: <filtered>\ndate: <filtered>\nserver: <filtered>\ncache-control: <filtered>\nx-xss-protection: <filtered>\nx-frame-options: <filtered>\nx-content-type-options: <filtered>\nalt-svc: <filtered>\naccept-ranges: <filtered>\nx-http2-stream-id: <filtered>\ntransfer-encoding: <filtered>","logger_name":"reactor.netty.http.client.HttpClientOperations","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.521-05:00","@version":"1","message":"[f64d0420/1-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] [terminated=false, cancelled=false, pending=0, error=null]: subscribing inbound receiver","logger_name":"reactor.netty.channel.FluxReceive","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.521-05:00","@version":"1","message":"[f64d0420/1-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Received last HTTP packet","logger_name":"reactor.netty.http.client.HttpClientOperations","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.522-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443](H2 - 3) Removed handler: reactor.left.responseTimeoutHandler, pipeline: AbstractHttp2StreamChannel$3{(reactor.left.h2ToHttp11Codec = io.netty.handler.codec.http2.Http2StreamFrameToHttpObjectCodec), (reactor.left.httpTrafficHandler = reactor.netty.http.client.Http2StreamBridgeClientHandler), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}","logger_name":"reactor.netty.ReactorNetty","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.522-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443](H2 - 3) Stream closed, now: 0 active streams and 100 max active streams.","logger_name":"reactor.netty.http.client.Http2ConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.522-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Channel activated","logger_name":"reactor.netty.http.client.Http2Pool","thread_name":"consumer-12-C-1","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.522-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Channel deactivated","logger_name":"reactor.netty.http.client.Http2Pool","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.522-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443](H2 - -1) New HTTP/2 stream","logger_name":"reactor.netty.http.client.HttpClientOperations","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.522-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443](H2 - -1) Initialized HTTP/2 stream pipeline AbstractHttp2StreamChannel$3{(reactor.left.h2ToHttp11Codec = io.netty.handler.codec.http2.Http2StreamFrameToHttpObjectCodec), (reactor.left.httpTrafficHandler = reactor.netty.http.client.Http2StreamBridgeClientHandler), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}","logger_name":"reactor.netty.http.client.HttpClientConfig","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.522-05:00","@version":"1","message":"[f64d0420/2-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Handler is being applied: {uri=https://fcm.googleapis.com/v1/projects/wmt-sumo-pnp-poc/messages:send, method=POST}","logger_name":"reactor.netty.http.client.HttpClientConnect","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.523-05:00","@version":"1","message":"[f64d0420/2-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Stream opened, now: 1 active streams and 100 max active streams.","logger_name":"reactor.netty.http.client.Http2ConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.523-05:00","@version":"1","message":"[f64d0420/2-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Added encoder [reactor.left.responseTimeoutHandler] at the beginning of the user pipeline, full pipeline: [reactor.left.h2ToHttp11Codec, reactor.left.httpTrafficHandler, reactor.left.responseTimeoutHandler, reactor.right.reactiveBridge, DefaultChannelPipeline$TailContext#0]","logger_name":"reactor.netty.ReactorNetty","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.629-05:00","@version":"1","message":"[f64d0420/2-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Received response (auto-read:false) : RESPONSE(decodeResult: success, version: HTTP/1.1)\nHTTP/1.1 403 Forbidden\nvary: <filtered>\nvary: <filtered>\nvary: <filtered>\ncontent-type: <filtered>\ndate: <filtered>\nserver: <filtered>\ncache-control: <filtered>\nx-xss-protection: <filtered>\nx-frame-options: <filtered>\nx-content-type-options: <filtered>\nalt-svc: <filtered>\naccept-ranges: <filtered>\nx-http2-stream-id: <filtered>\ntransfer-encoding: <filtered>","logger_name":"reactor.netty.http.client.HttpClientOperations","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.629-05:00","@version":"1","message":"[f64d0420/2-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] [terminated=false, cancelled=false, pending=0, error=null]: subscribing inbound receiver","logger_name":"reactor.netty.channel.FluxReceive","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.629-05:00","@version":"1","message":"[f64d0420/2-1, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443] Received last HTTP packet","logger_name":"reactor.netty.http.client.HttpClientOperations","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.630-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443](H2 - 5) Removed handler: reactor.left.responseTimeoutHandler, pipeline: AbstractHttp2StreamChannel$3{(reactor.left.h2ToHttp11Codec = io.netty.handler.codec.http2.Http2StreamFrameToHttpObjectCodec), (reactor.left.httpTrafficHandler = reactor.netty.http.client.Http2StreamBridgeClientHandler), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}","logger_name":"reactor.netty.ReactorNetty","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}
{"@timestamp":"2024-04-18T19:24:31.630-05:00","@version":"1","message":"[f64d0420, L:/10.225.123.154:52511 - R:fcm.googleapis.com/216.239.36.57:443](H2 - 5) Stream closed, now: 0 active streams and 100 max active streams.","logger_name":"reactor.netty.http.client.Http2ConnectionProvider","thread_name":"reactor-http-nio-3","level":"DEBUG","level_value":10000}

Possible Solution

Your Environment

Spring boot 2.7.12

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

• Reactor version(s) used:
• Other relevant libraries versions (eg. netty, ...):
• JVM version (java -version): Java 11
• OS and version (eg. uname -a):

[violetagg]

@KrishnakanthYachareni From the configuration I don't see any changes related to the memory management, which means that you are running with Pooled ByteBufs. Is it possible that you see this reserved memory?
Check this https://www.programmersought.com/article/9322400832/ ,
https://netty.io/wiki/reference-counted-objects.html and may be this https://projectreactor.io/docs/netty/release/reference/index.html#faq.memory-leaks

[KrishnakanthYachareni]

@violetagg Yes, my application has been using default Pooled ByteBus, as I did not have any manual configuration related to mememory management. You can see below it is a reserved memory.

If you see the memory usage of the pod which is gradually increasing in the pod and memory is not collecting by GC.

Why does ByteBuf memory is not clearing, is the following snippet not releasing the buffer properly?

Flux<FcmResponseWrapper> response = fcmRestClient.sendPush(fluxRequests, appName, eventIdentifiers);

List<Tuple2<Long, FcmResponseWrapper>> listOfSendResponses = response.elapsed()
                  .collectList()
                  .block(); // Thread blocking.

[violetagg]

Is it possible to enable memory metrics provided by Reactor Netty
https://projectreactor.io/docs/netty/release/reference/index.html#_metrics_6
reactor.netty.bytebuf.allocator.active.heap.memory
reactor.netty.bytebuf.allocator.active.direct.memory

[KrishnakanthYachareni]

@violetagg Just to clarify that in my code no other objects are allocating more memory and memeory is leaking from reactor related objects. That I verfied during JVM profiling.

bytebuf are not emitted but ConnectionProvider metrics are emitting. Following code used to enable the reactor metrics. am I missing anything?

Schedulers.enableMetrics(); // in Spring boot main method.

private ConnectionProvider getConnectionProvider() {
        return ConnectionProvider.builder("fcmConnection")
                .metrics(true)
                .build();
    }

    @Bean
    public NettyServerCustomizer nettyServerCustomizer(){
        return httpServer -> httpServer.metrics(true, Function.identity());
    }

My suspect is .subscribeOn(Schedulers.fromExecutor(executor) method with reactor scheduler may causing the leak, Can you please validate the code?

messages.flatMap(msgPair -> webClient.post()
                            .uri(appURI)
                            .header(HttpHeaders.AUTHORIZATION, accessToken)
                            .body(Mono.just(msgPair.getFirst()), FirebaseRequest.class)
                            .exchangeToFlux(clientResponse -> this.handleResponse(clientResponse, msgPair.getSecond())))
                    .subscribeOn(Schedulers.fromExecutor(executor));

[violetagg]

Add this to the client httpClient.metrics(true, Function.identity());

[KrishnakanthYachareni]

@violetagg I collected the byteBuff memory metric and both seems always constant and same value.

[violetagg]

@KrishnakanthYachareni You need the active not the overal memory that is reserved

reactor.netty.bytebuf.allocator.active.heap.memory
reactor.netty.bytebuf.allocator.active.direct.memory

vs

reactor.netty.bytebuf.allocator.used.heap.memory
reactor.netty.bytebuf.allocator.used.direct.memory

[KrishnakanthYachareni]

@violetagg My bad, I miss understood the mertics name. Both memories are most of the time showing as 0 but some time direct memory have certain value and active heap memory is always 0. what does it mean?

Note: Just to mention Non-Heap memory is gradually increasing and not going down.

[KrishnakanthYachareni]

Took this screenshot after few hours of running and later again both memories went down to 0

[violetagg]

@KrishnakanthYachareni So to summarise the active memory is not increasing constantly but it returns to 0?

[KrishnakanthYachareni]

@violetagg Yes. that is right!

[violetagg]

@KrishnakanthYachareni So looks like no memory leaks?

[KrishnakanthYachareni]

@violetagg I did these changes, this could be a reason memory leak got fixed.

From

webClient.subscribeOn(Schedulers.fromExecutor(executor));

To

webClient.publishOn(Schedulers.fromExecutor(executor));

[violetagg]

@KrishnakanthYachareni I cannot comment on the change from subscribeOn/publishOn without having a reproducible example. That said should I close this, if the above is working for you?

[KrishnakanthYachareni]

@violetagg yes please, you can close it.