Repository: netty/netty
Issue: DNS Resolver leaking direct memory
Labels: defect

### Expected behavior

DNS resolver does not leak direct memory when buffers are freed correctly.

### Actual behavior

Memory is leaked, JVM runs out of direct memory.

```
[error] Exception in thread "main" io.netty.resolver.dns.DnsNameResolverException: [/10.0.2.3:53] failed to send a query via UDP (no stack trace available)
[error] Caused by: io.netty.handler.codec.EncoderException: io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 16777216 byte(s) of direct memory (used: 234881024, max: 239075328)
```
full stacktrace: https://gist.github.com/dziemba/ffeb8f2c3b131a74c042746151ac05c0

### Steps to reproduce

- Use `DnsNameResolver.resolveAll` to query SRV records for a name that contains a lot of answers (e.g. `hugedns.test.dziemba.net`)
-  Run query repeatedly and `.release()` returned DNS records. 
- Watch java process memory usage increase until it finally throws with `io.netty.util.internal.OutOfDirectMemoryError`

It might be important that the DNS response is large so it gets truncated and then retried via TCP.

### Minimal yet complete reproducer code (or URL to code)

https://gist.github.com/dziemba/c904d227d105b6fc7cf00495257fbb40
- run with `-Xmx128m -XX:MaxDirectMemorySize=4m` to make it fail quickly
- will fail as described after around 100-200 iterations (with above settings)
- sometimes it fails with actual network errors, re-run a few times if that happens

### Netty version

`4.1.42.Final`

### JVM version (e.g. `java -version`)
```
openjdk version "1.8.0_222"
OpenJDK Runtime Environment (AdoptOpenJDK)(build 1.8.0_222-b10)
OpenJDK 64-Bit Server VM (AdoptOpenJDK)(build 25.222-b10, mixed mode)
```

I was not able to reproduce this on Java 11 or 12.

### OS version (e.g. `uname -a`)

```
Darwin wopro3 18.7.0 Darwin Kernel Version 18.7.0: Tue Aug 20 16:57:14 PDT 2019; root:xnu-4903.271.2~2/RELEASE_X86_64 x86_64
```

I also tried to reproduce this on Linux. It shows the same error behavior (fails after 100 runs of the test script) but does not output the `OutOfDirectMemoryError` for some reason...

Issue comments:
@dziemba could you also try running it with `-Dio.netty.leakDetection.level=paranoid`?
@normanmaurer I think the problem might be here: https://github.com/netty/netty/blob/68673b652e422df6db462afb8c8f3ce9413c5c90/resolver-dns/src/main/java/io/netty/resolver/dns/DnsNameResolver.java#L1219

Unless I'm missing something, it doesn't look like there's any release corresponding to this retain in the later callbacks. Also it would in any case be better to replace this line with just `msg = null;` - no need to retain if releasing immediately after.

Something like this, though maybe with some more try/finallys: 82da56c77f90c572f0dbbfcb408c780e506d42ca
@njhill Here's a complete debug log with paranoid leak detection:

https://gist.github.com/dziemba/b55e1a5bea461d9ac6114cc0ae74b078
@njhill @dziemba imho the retain() is correct here. I will need to look into this in more detail.
@normanmaurer apologies I was kind of conflating two different observations. I agree the retain() as currently stands is correct, it could just be replaced by `msg = null` as a small optimization (this is separate to the problem at hand). My other point was that there may not always be a corresponding release() but now agree that's not what's happening in this case.

I looked a bit closer and think I found the issue(s)... will open a PR soon.
@njhill ok cool... Will not spend more time on this then and just wait for your PR :)
@normanmaurer @dziemba please see #9479 

FYI the main leak in the current code is here - the result of the future needs to be released, since the `finish` method that it's passed to does not take ref-count ownership: https://github.com/netty/netty/blob/4dc1eccf60252f8a690610c20495ed095e274d3a/resolver-dns/src/main/java/io/netty/resolver/dns/DnsNameResolver.java#L1302
Thank you for fixing this! ❤️ 
@dziemba no problem, thank you for reporting it with so much detail

Pull request review comments:
@njhill can we also verify that this line returns false (add a `assertFalse`):
https://github.com/netty/netty/blob/a95d04bb2f6b5f9d000f22b4daa36b5b1f12935b/resolver-dns/src/test/java/io/netty/resolver/dns/DnsNameResolverTest.java#L2901
@normanmaurer do you mean that it should return true (no more refs left)? I just tried with false and it fails :)
yes sorry... `true` :)
nit: release first before log. 
I think we should preserve the logic and fallback to the original truncated response if we did not receive a TCP response just like we did before:

https://github.com/netty/netty/pull/9647/files#diff-6678a05ef56f2434e7aa20acbe9d97e9L1277

Also beside this we should not log the exception if we received the response before. As the exception may be just a "connection reset" and so harmless. Also consider calling `ctx.close()` in any case to ensure the channel is closed in all cases. 


why not make this a real java doc ?
lets rename this to `trySuccess` as this is what it does ;) 
Now fixed. I didn’t add an explicit close since the channel is closed immediately in the listener below.
nit: why not add these also in the `ChannelInitializer` above ?
@normanmaurer the latter one can't be done there since it references `tcpCtx` which is only created post-connect. But also none of the handlers are needed anyhow unless/until the channel connects successfully. I actually first tried to remove the `ChannelInitializer` completely, but `Bootstrap` requires at least one handler to be provided - maybe we could reconsider that requirement?

I did just realize though that the `ChannelInitializer` can be replaced with the single handler that it adds - a small simplification. I just pushed another commit to do that.

I'm also fine with adding it back along with the first two handlers if you would prefer that, but the last one needs to be here I think if we are to keep the query context verification.

Patch that resolved the issue:
diff --git a/resolver-dns/src/main/java/io/netty/resolver/dns/DnsNameResolver.java b/resolver-dns/src/main/java/io/netty/resolver/dns/DnsNameResolver.java
index 4e84a475f81..c7551d8056c 100644
--- a/resolver-dns/src/main/java/io/netty/resolver/dns/DnsNameResolver.java
+++ b/resolver-dns/src/main/java/io/netty/resolver/dns/DnsNameResolver.java
@@ -50,7 +50,6 @@
 import io.netty.resolver.InetNameResolver;
 import io.netty.resolver.ResolvedAddressTypes;
 import io.netty.util.NetUtil;
-import io.netty.util.ReferenceCountUtil;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.FastThreadLocal;
 import io.netty.util.concurrent.Future;
@@ -1199,129 +1198,107 @@ private final class DnsResponseHandler extends ChannelInboundHandlerAdapter {
 
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object msg) {
-            try {
-                final DatagramDnsResponse res = (DatagramDnsResponse) msg;
-                final int queryId = res.id();
+            final DatagramDnsResponse res = (DatagramDnsResponse) msg;
+            final int queryId = res.id();
 
-                if (logger.isDebugEnabled()) {
-                    logger.debug("{} RECEIVED: UDP [{}: {}], {}", ch, queryId, res.sender(), res);
-                }
+            if (logger.isDebugEnabled()) {
+                logger.debug("{} RECEIVED: UDP [{}: {}], {}", ch, queryId, res.sender(), res);
+            }
 
-                final DnsQueryContext qCtx = queryContextManager.get(res.sender(), queryId);
-                if (qCtx == null) {
-                    logger.warn("{} Received a DNS response with an unknown ID: {}", ch, queryId);
-                    return;
-                }
+            final DnsQueryContext qCtx = queryContextManager.get(res.sender(), queryId);
+            if (qCtx == null) {
+                logger.warn("{} Received a DNS response with an unknown ID: {}", ch, queryId);
+                res.release();
+                return;
+            }
+
+            // Check if the response was truncated and if we can fallback to TCP to retry.
+            if (!res.isTruncated() || socketChannelFactory == null) {
+                qCtx.finish(res);
+                return;
+            }
 
-                // Check if the response was truncated and if we can fallback to TCP to retry.
-                if (res.isTruncated() && socketChannelFactory != null) {
-                    // Let's retain as we may need it later on.
-                    res.retain();
-
-                    Bootstrap bs = new Bootstrap();
-                    bs.option(ChannelOption.SO_REUSEADDR, true)
-                            .group(executor())
-                            .channelFactory(socketChannelFactory)
-                            .handler(new ChannelInitializer<Channel>() {
-                                @Override
-                                protected void initChannel(Channel ch) {
-                                    ch.pipeline().addLast(TCP_ENCODER);
-                                    ch.pipeline().addLast(new TcpDnsResponseDecoder());
-                                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
-                                        private boolean finish;
-
-                                        @Override
-                                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
-                                            try {
-                                                Channel channel = ctx.channel();
-                                                DnsResponse response = (DnsResponse) msg;
-                                                int queryId = response.id();
-
-                                                if (logger.isDebugEnabled()) {
-                                                    logger.debug("{} RECEIVED: TCP [{}: {}], {}", channel, queryId,
-                                                            channel.remoteAddress(), response);
-                                                }
-
-                                                DnsQueryContext tcpCtx = queryContextManager.get(res.sender(), queryId);
-                                                if (tcpCtx == null) {
-                                                    logger.warn("{} Received a DNS response with an unknown ID: {}",
-                                                            channel, queryId);
-                                                    qCtx.finish(res);
-                                                    return;
-                                                }
-
-                                                // Release the original response as we will use the response that we
-                                                // received via TCP fallback.
-                                                res.release();
-
-                                                tcpCtx.finish(new AddressedEnvelopeAdapter(
-                                                        (InetSocketAddress) ctx.channel().remoteAddress(),
-                                                        (InetSocketAddress) ctx.channel().localAddress(),
-                                                        response));
-
-                                                finish = true;
-                                            } finally {
-                                                ReferenceCountUtil.release(msg);
-                                            }
-                                        }
-
-                                        @Override
-                                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-                                            if (!finish) {
-                                                if (logger.isDebugEnabled()) {
-                                                    logger.debug("{} Error during processing response: TCP [{}: {}]",
-                                                            ctx.channel(), queryId,
-                                                            ctx.channel().remoteAddress(), cause);
-                                                }
-                                                // TCP fallback failed, just use the truncated response as
-                                                qCtx.finish(res);
-                                            }
-                                        }
-                                    });
-                                }
-                            });
-                    bs.connect(res.sender()).addListener(new ChannelFutureListener() {
+            Bootstrap bs = new Bootstrap();
+            bs.option(ChannelOption.SO_REUSEADDR, true)
+            .group(executor())
+            .channelFactory(socketChannelFactory)
+            .handler(TCP_ENCODER);
+            bs.connect(res.sender()).addListener(new ChannelFutureListener() {
+                @Override
+                public void operationComplete(ChannelFuture future) {
+                    if (!future.isSuccess()) {
+                        if (logger.isDebugEnabled()) {
+                            logger.debug("{} Unable to fallback to TCP [{}]", queryId, future.cause());
+                        }
+
+                        // TCP fallback failed, just use the truncated response.
+                        qCtx.finish(res);
+                        return;
+                    }
+                    final Channel channel = future.channel();
+
+                    Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise =
+                            channel.eventLoop().newPromise();
+                    final TcpDnsQueryContext tcpCtx = new TcpDnsQueryContext(DnsNameResolver.this, channel,
+                            (InetSocketAddress) channel.remoteAddress(), qCtx.question(),
+                            EMPTY_ADDITIONALS, promise);
+
+                    channel.pipeline().addLast(new TcpDnsResponseDecoder());
+                    channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                         @Override
-                        public void operationComplete(ChannelFuture future) {
-                            if (future.isSuccess()) {
-                                final Channel channel = future.channel();
-
-                                Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise =
-                                        channel.eventLoop().newPromise();
-                                new TcpDnsQueryContext(DnsNameResolver.this, channel,
-                                        (InetSocketAddress) channel.remoteAddress(), qCtx.question(),
-                                        EMPTY_ADDITIONALS, promise).query(true, future.channel().newPromise());
-                                promise.addListener(
-                                        new FutureListener<AddressedEnvelope<DnsResponse, InetSocketAddress>>() {
-                                    @Override
-                                    public void operationComplete(
-                                            Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> future) {
-                                        channel.close();
-
-                                        if (future.isSuccess()) {
-                                            qCtx.finish(future.getNow());
-                                        } else {
-                                            // TCP fallback failed, just use the truncated response.
-                                            qCtx.finish(res);
-                                        }
-                                    }
-                                });
+                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                            Channel channel = ctx.channel();
+                            DnsResponse response = (DnsResponse) msg;
+                            int queryId = response.id();
+
+                            if (logger.isDebugEnabled()) {
+                                logger.debug("{} RECEIVED: TCP [{}: {}], {}", channel, queryId,
+                                        channel.remoteAddress(), response);
+                            }
+
+                            DnsQueryContext foundCtx = queryContextManager.get(res.sender(), queryId);
+                            if (foundCtx == tcpCtx) {
+                                tcpCtx.finish(new AddressedEnvelopeAdapter(
+                                        (InetSocketAddress) ctx.channel().remoteAddress(),
+                                        (InetSocketAddress) ctx.channel().localAddress(),
+                                        response));
                             } else {
-                                if (logger.isDebugEnabled()) {
-                                    logger.debug("{} Unable to fallback to TCP [{}]", queryId, future.cause());
-                                }
+                                response.release();
+                                tcpCtx.tryFailure("Received TCP response with unexpected ID", null, false);
+                                logger.warn("{} Received a DNS response with an unexpected ID: {}",
+                                        channel, queryId);
+                            }
+                        }
+
+                        @Override
+                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+                            if (tcpCtx.tryFailure("TCP fallback error", cause, false) && logger.isDebugEnabled()) {
+                                logger.debug("{} Error during processing response: TCP [{}: {}]",
+                                        ctx.channel(), queryId,
+                                        ctx.channel().remoteAddress(), cause);
+                            }
+                        }
+                    });
 
+                    promise.addListener(
+                            new FutureListener<AddressedEnvelope<DnsResponse, InetSocketAddress>>() {
+                        @Override
+                        public void operationComplete(
+                                Future<AddressedEnvelope<DnsResponse, InetSocketAddress>> future) {
+                            channel.close();
+
+                            if (future.isSuccess()) {
+                                qCtx.finish(future.getNow());
+                                res.release();
+                            } else {
                                 // TCP fallback failed, just use the truncated response.
                                 qCtx.finish(res);
                             }
                         }
                     });
-                } else {
-                    qCtx.finish(res);
+                    tcpCtx.query(true, future.channel().newPromise());
                 }
-            } finally {
-                ReferenceCountUtil.safeRelease(msg);
-            }
+            });
         }
 
         @Override
diff --git a/resolver-dns/src/main/java/io/netty/resolver/dns/DnsQueryContext.java b/resolver-dns/src/main/java/io/netty/resolver/dns/DnsQueryContext.java
index d5ba91e5e85..d476c1f2fe8 100644
--- a/resolver-dns/src/main/java/io/netty/resolver/dns/DnsQueryContext.java
+++ b/resolver-dns/src/main/java/io/netty/resolver/dns/DnsQueryContext.java
@@ -159,7 +159,7 @@ public void operationComplete(ChannelFuture future) {
 
     private void onQueryWriteCompletion(ChannelFuture writeFuture) {
         if (!writeFuture.isSuccess()) {
-            setFailure("failed to send a query via " + protocol(), writeFuture.cause());
+            tryFailure("failed to send a query via " + protocol(), writeFuture.cause(), false);
             return;
         }
 
@@ -174,40 +174,37 @@ public void run() {
                         return;
                     }
 
-                    setFailure("query via " + protocol() + " timed out after " +
-                            queryTimeoutMillis + " milliseconds", null);
+                    tryFailure("query via " + protocol() + " timed out after " +
+                            queryTimeoutMillis + " milliseconds", null, true);
                 }
             }, queryTimeoutMillis, TimeUnit.MILLISECONDS);
         }
     }
 
+    /**
+     * Takes ownership of passed envelope
+     */
     void finish(AddressedEnvelope<? extends DnsResponse, InetSocketAddress> envelope) {
         final DnsResponse res = envelope.content();
         if (res.count(DnsSection.QUESTION) != 1) {
             logger.warn("Received a DNS response with invalid number of questions: {}", envelope);
-            return;
-        }
-
-        if (!question().equals(res.recordAt(DnsSection.QUESTION))) {
+        } else if (!question().equals(res.recordAt(DnsSection.QUESTION))) {
             logger.warn("Received a mismatching DNS response: {}", envelope);
-            return;
+        } else if (trySuccess(envelope)) {
+            return; // Ownership transferred, don't release
         }
-
-        setSuccess(envelope);
+        envelope.release();
     }
 
-    private void setSuccess(AddressedEnvelope<? extends DnsResponse, InetSocketAddress> envelope) {
-        Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise = this.promise;
-        @SuppressWarnings("unchecked")
-        AddressedEnvelope<DnsResponse, InetSocketAddress> castResponse =
-                (AddressedEnvelope<DnsResponse, InetSocketAddress>) envelope.retain();
-        if (!promise.trySuccess(castResponse)) {
-            // We failed to notify the promise as it was failed before, thus we need to release the envelope
-            envelope.release();
-        }
+    @SuppressWarnings("unchecked")
+    private boolean trySuccess(AddressedEnvelope<? extends DnsResponse, InetSocketAddress> envelope) {
+        return promise.trySuccess((AddressedEnvelope<DnsResponse, InetSocketAddress>) envelope);
     }
 
-    private void setFailure(String message, Throwable cause) {
+    boolean tryFailure(String message, Throwable cause, boolean timeout) {
+        if (promise.isDone()) {
+            return false;
+        }
         final InetSocketAddress nameServerAddr = nameServerAddr();
 
         final StringBuilder buf = new StringBuilder(message.length() + 64);
@@ -218,14 +215,14 @@ private void setFailure(String message, Throwable cause) {
            .append(" (no stack trace available)");
 
         final DnsNameResolverException e;
-        if (cause == null) {
+        if (timeout) {
             // This was caused by an timeout so use DnsNameResolverTimeoutException to allow the user to
             // handle it special (like retry the query).
             e = new DnsNameResolverTimeoutException(nameServerAddr, question(), buf.toString());
         } else {
             e = new DnsNameResolverException(nameServerAddr, question(), buf.toString(), cause);
         }
-        promise.tryFailure(e);
+        return promise.tryFailure(e);
     }
 
     @Override


Test patch:
diff --git a/resolver-dns/src/test/java/io/netty/resolver/dns/DnsNameResolverTest.java b/resolver-dns/src/test/java/io/netty/resolver/dns/DnsNameResolverTest.java
index 38a27ca4a1d..3e35e288a5b 100644
--- a/resolver-dns/src/test/java/io/netty/resolver/dns/DnsNameResolverTest.java
+++ b/resolver-dns/src/test/java/io/netty/resolver/dns/DnsNameResolverTest.java
@@ -71,6 +71,7 @@
 import org.junit.rules.ExpectedException;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.DatagramSocket;
 import java.net.Inet4Address;
 import java.net.InetAddress;
@@ -2759,6 +2760,25 @@ public void testTruncatedWithTcpFallbackBecauseOfMtu() throws IOException {
         testTruncated0(true, true);
     }
 
+    private static DnsMessageModifier modifierFrom(DnsMessage message) {
+        DnsMessageModifier modifier = new DnsMessageModifier();
+        modifier.setAcceptNonAuthenticatedData(message.isAcceptNonAuthenticatedData());
+        modifier.setAdditionalRecords(message.getAdditionalRecords());
+        modifier.setAnswerRecords(message.getAnswerRecords());
+        modifier.setAuthoritativeAnswer(message.isAuthoritativeAnswer());
+        modifier.setAuthorityRecords(message.getAuthorityRecords());
+        modifier.setMessageType(message.getMessageType());
+        modifier.setOpCode(message.getOpCode());
+        modifier.setQuestionRecords(message.getQuestionRecords());
+        modifier.setRecursionAvailable(message.isRecursionAvailable());
+        modifier.setRecursionDesired(message.isRecursionDesired());
+        modifier.setReserved(message.isReserved());
+        modifier.setResponseCode(message.getResponseCode());
+        modifier.setTransactionId(message.getTransactionId());
+        modifier.setTruncated(message.isTruncated());
+        return modifier;
+    }
+
     private static void testTruncated0(boolean tcpFallback, final boolean truncatedBecauseOfMtu) throws IOException {
         final String host = "somehost.netty.io";
         final String txt = "this is a txt record";
@@ -2784,20 +2804,7 @@ protected DnsMessage filterMessage(DnsMessage message) {
 
                 if (!truncatedBecauseOfMtu) {
                     // Create a copy of the message but set the truncated flag.
-                    DnsMessageModifier modifier = new DnsMessageModifier();
-                    modifier.setAcceptNonAuthenticatedData(message.isAcceptNonAuthenticatedData());
-                    modifier.setAdditionalRecords(message.getAdditionalRecords());
-                    modifier.setAnswerRecords(message.getAnswerRecords());
-                    modifier.setAuthoritativeAnswer(message.isAuthoritativeAnswer());
-                    modifier.setAuthorityRecords(message.getAuthorityRecords());
-                    modifier.setMessageType(message.getMessageType());
-                    modifier.setOpCode(message.getOpCode());
-                    modifier.setQuestionRecords(message.getQuestionRecords());
-                    modifier.setRecursionAvailable(message.isRecursionAvailable());
-                    modifier.setRecursionDesired(message.isRecursionDesired());
-                    modifier.setReserved(message.isReserved());
-                    modifier.setResponseCode(message.getResponseCode());
-                    modifier.setTransactionId(message.getTransactionId());
+                    DnsMessageModifier modifier = modifierFrom(message);
                     modifier.setTruncated(true);
                     return modifier.getDnsMessage();
                 }
@@ -2842,8 +2849,15 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
                 // If we are configured to use TCP as a fallback lets replay the dns message over TCP
                 Socket socket = serverSocket.accept();
 
+                InputStream in = socket.getInputStream();
+                assertTrue((in.read() << 8 | (in.read() & 0xff)) > 2); // skip length field
+                int txnId = in.read() << 8 | (in.read() & 0xff);
+
                 IoBuffer ioBuffer = IoBuffer.allocate(1024);
-                new DnsMessageEncoder().encode(ioBuffer, messageRef.get());
+                // Must replace the transactionId with the one from the TCP request
+                DnsMessageModifier modifier = modifierFrom(messageRef.get());
+                modifier.setTransactionId(txnId);
+                new DnsMessageEncoder().encode(ioBuffer, modifier.getDnsMessage());
                 ioBuffer.flip();
 
                 ByteBuffer lenBuffer = ByteBuffer.allocate(2);
@@ -2884,7 +2898,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
             } else {
                 assertTrue(envelope.content().isTruncated());
             }
-            envelope.release();
+            assertTrue(envelope.release());
         } finally {
             dnsServer2.stop();
             if (resolver != null) {