Repository: netty/netty (Java)

Issue:
DNS Resolver leaking direct memory
### 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...


Hints:
@dziemba could you also try running it with `-Dio.netty.leakDetection.level=paranoid`?

Patch:
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
--- 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
--- 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
--- 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) {