Direct Memory grows from 5MB-1GB leading to application crash

[abhishek-sharma-20]

We have a routing reactive application which take a request from client and then routes this request to a backend. We originally had a non-reactive solution now we are moving to reactive solution. We have a flow in which backend response is large (1MB-5MB) in these cases when running with 2-3TPS for a duration of 30mins to 1 hour we see application getting restarted automatically . On investigating we saw it is getting restarted because application is consuming more memory and hitting memory limit of pod and leading to restart. On observing memory usage patterns we saw direct memory was keep on growing from 5MB - 1024 MB and then leading to application restart.

Expected Behavior

Application should be able to support large response size and should not consume more memory than expected.

Actual Behavior

Application is consuming more memory than expected and leading to application restart.

Steps to Reproduce

###HttpConfig.java

@Bean
    public ConnectionProvider connectionProvider() {
        return ConnectionProvider.builder("httpConn")
                .maxConnections(20)
                .metrics(true)
                .build();
    }

    @Bean
    LoopResources loopResources() {
        return LoopResources.create("loop", 100, true);
    }

    @Bean
    public PooledByteBufAllocator byteBufAllocator() {
        return PooledByteBufAllocator.DEFAULT;
    }

    public SslContext armSSLContext() {
        SslContext sslContext = null;
        String keyStoreFile = System.getProperty("javax.net.ssl.keyStore");
        char[] keyStorePassword = System.getProperty("javax.net.ssl.keyStorePassword").toCharArray();
        try (InputStream keyStoreStream = Files.newInputStream(Paths.get(keyStoreFile))) {
            KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
            keyStore.load(keyStoreStream, keyStorePassword);
            KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance("SunX509");
            keyManagerFactory.init(keyStore, keyStorePassword);

            KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
            trustStore.load(new FileInputStream(keyStoreFile), keyStorePassword);

            TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("SunX509");
            trustManagerFactory.init(trustStore);

            sslContext = SslContextBuilder.forClient().keyManager(keyManagerFactory)
                    .trustManager(trustManagerFactory)
                    .build();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return sslContext;
    }

###HttpWebClient.java

 public WebClient reactiveWebClientBuilder() {
        HttpClient httpClient = HttpClient.create(connectionProvider)
                .resolver(DefaultAddressResolverGroup.INSTANCE)
                .secure(spec -> spec.sslContext(armSSLContext))
                .compress(true)
                .metrics(true, s->s)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 0)
                .option(ChannelOption.TCP_NODELAY, true)
                .doOnConnected(conn -> conn.addHandlerLast(new ReadTimeoutHandler(
                        (int) TimeUnit.MILLISECONDS.toSeconds(connectionTimeout.intValue())
                )))
                .runOn(loopResources);
        return WebClient.builder()
            .clientConnector(new ReactorClientHttpConnector(httpClient))
            .codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(armProperties.getHttp().getByteBufferSize() *1024 * 1024)) // 6MB
            .build();
    }

###ResponseController

 @GetMapping("/response")
    public Mono<ResponseEntity<String>> getResponse(@RequestParam int size) {
        return reactiveWebClientBuilder.method(HttpMethod.GET)
                .uri(uriBuilder -> uriBuilder.scheme("https").host(armProp.getHttp().getHostname())
            .port(armProp.getHttp().getPort()).path("/path").build())
                .contentType(MediaType.APPLICATION_JSON)
                .headers(this::addHttpHeaders).retrieve()
                .onStatus(httpStatusCode -> HttpStatus.valueOf(httpStatusCode.value())
                        .isError(), t -> Mono.empty()).toEntity(String.class);
    }

Above are our classes used for this app and below is JVM arguments used for this application
-Xmx3072M -Xss256K -Xms3072M -XX:+HeapDumpOnOutOfMemoryError -verbose:gc -XX:+ParallelRefProcEnabled -XX:ParallelGCThreads=4 -XX:ConcGCThreads=2 -XX:MaxGCPauseMillis=500 -XX:MetaspaceSize=64m -XX:MaxMetaspaceSize=512m -XX:+DisableExplicitGC -XX:MaxJavaStackTraceDepth=15 -Dspring.config.location=optional:classpath:/,optional:classpath:/config/ -Dorg.springframework.boot.logging.LoggingSystem=none

Possible Solution

Your Environment

• reactor-netty-core: 1.2.0
• netty: 4.1.111.Final
• spring-framework: 6.1.13
• spring-boot: 3.2.9

Below are system details

• System Linux (5.4.0-200-generic)
• OpenJDK 17.0.13
• Kubernetes pod with 3 core CPU and 4 GB RAM

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