reactor.netty.ReactorNetty$InternalNettyException: io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 16777216 byte(s) of direct memory

Question

Software versions in use: spring-webflux-5.3.4, reactor-core-3.4.4, spring-data-mongodb-3.1.6

• Am building a spring boot application that uses spring webclient to invoke an image service that will serve a pdf image back.
• The returned pdf is then stored in mongodb using spring's ReactiveGridfsTemplate.
• For performance testing am having the service return 120 MB pdf all the time.
• First invocation of the service and storing the returned pdf in mongodb works fine and happens in under 10 seconds.
• However, second invocation onward, I start getting the following error while storing the returned pdf in mongodb. Can someone advise on what am doing wrong?

Caused by: io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 16777216 byte(s) of direct memory (used: 1056964615, max: 1073741824) at io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:776) at io.netty.util.internal.PlatformDependent.allocateDirectNoCleaner(PlatformDependent.java:731) at io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:645) at io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:621) at io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:204) at io.netty.buffer.PoolArena.tcacheAllocateNormal(PoolArena.java:188) at io.netty.buffer.PoolArena.allocate(PoolArena.java:138) at io.netty.buffer.PoolArena.allocate(PoolArena.java:128) at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:378) at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187) at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:178) at io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:139) at io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator$MaxMessageHandle.allocate(DefaultMaxMessagesRecvByteBufAllocator.java:114) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:150) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719) at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.lang.Thread.run(Thread.java:825)

Code to build webclient:

WebClient webClient = WebClient.builder().filter(WebClientFilter.logRequest())// for logging request
                .filter(WebClientFilter.logResponse()) // for logging response
                .exchangeStrategies(ExchangeStrategies.builder()
                        .codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(5242880)).build())
                .build();

Code to invoke image service using webclient:

Flux<DataBuffer> imageFlux = webClient.method(httpmethod).uri(uri)
                    .bodyValue((payloadBody == null) ? StringUtils.EMPTY : payloadBody.toPayloadBody())
                    .accept(MediaType.ALL).exchangeToFlux(response -> {
                        logger.log(Level.DEBUG, "DefaultHttpClient exchangeToFlux got response with status code {}",response.statusCode());
                        if (response.statusCode().is4xxClientError() || response.statusCode().is5xxServerError()) {
                            logger.log(Level.ERROR,
                                    "DefaultHttpClient exchangeToFlux encountered error {} throwing service exception",
                                    response.statusCode());
                            return Flux.error(new ServiceException(response.bodyToMono(String.class).flatMap(body -> {
                                return Mono.just(body);
                            }), response.rawStatusCode()));
                        }
    
                        return response.bodyToFlux(DataBuffer.class);
                    });

Code to store pdf in mongodb returned by image service using spring's ReactiveGridfsTemplate:

imageFlux is what I receive above.

protected Mono<ObjectId> getMono(Flux<DataBuffer> imageFlux , DocumentContext documentContext) {

    return reactiveGridFsTmpl.store(imageFlux, new java.util.Date() + ApplicationConstants.PDF_EXTENSION,
            <org.bson.Document object with attributes from application>);
            
}

Here's how am firing the store call by subscribing to Mono returned by getMono(....). Within onComplete and onError have tried to release data buffer

Mono<ObjectId> imageObjectId = getMono(imageFlux, documentContext);

            imageObjectId.subscribe(new Subscriber<ObjectId>() {

                @Override
                public void onComplete() {
                    logger.log(Level.DEBUG, SUBSCRIPTION_ON_COMPLETE);
                    DataBufferUtils.release(imageFlux.blockFirst()); --> Attempt to release databuffer
                    logger.log(Level.DEBUG, SUBSCRIPTION_ON_COMPLETE_RELEASE_DATABUFFER);
                }

                @Override
                public void onError(Throwable t) {
                    logger.log(Level.ERROR, SUBSCRIPTION_ON_ERROR + t);
                    if (t instanceof ServiceException) {
                        logger.log(Level.ERROR, "DocumentDao caught ServiceException.");
                        flagErrorRecord((ServiceException) t, documentContext);
                    }
                    DataBufferUtils.release(imageFlux.blockFirst()); --> Attempt to release databuffer
                    logger.log(Level.ERROR, SUBSCRIPTION_ON_ERROR_RELEASE_DATABUFFER);
                }

                @Override
                public void onNext(ObjectId t) {
                    logger.log(Level.DEBUG, SUBSCRIPTION_ON_NEXT + t.toString());
                }

                @Override
                public void onSubscribe(Subscription s) {
                    logger.log(Level.DEBUG, SUBSCRIPTION_ON_SUBSCRIBE);
                    s.request(1);
                }
            });

Answer 1

try to change the directMemory using the JAVA_OPTS Environment variable.

JBP_CONFIG_JAVA_OPTS: '{ java_opts: "-XX:MaxDirectMemorySize=2048m" }'

I see that 1G is not sufficient. so try to set it at 2G