netty: Fix getAttributes() data races in NettyClientTransportTest

Since approximately the LBv2 API (the current API) was introduced, gRPC
won't use a transport until it is ready. Long ago, transports could be
used before they were ready and these old tests were not waiting for the
negotiator to complete before starting. We need them to wait for the
handshake to complete to avoid a test-only data race in getAttributes()
noticed by TSAN.

Throwing away data frames in the Noop handshaker is necessary to act
like a normal handshaker; they don't allow data frames to pass until the
handshake is complete. Without the handling, it goes through invalid
code paths in NettyClientHandler where a terminated transport becomes
ready, and a similar data race.

```
  Write of size 4 at 0x00008db31e2c by thread T37:
    #0 io.grpc.netty.NettyClientHandler.handleProtocolNegotiationCompleted(Lio/grpc/Attributes;Lio/grpc/InternalChannelz$Security;)V NettyClientHandler.java:517
    #1 io.grpc.netty.ProtocolNegotiators$GrpcNegotiationHandler.userEventTriggered(Lio/netty/channel/ChannelHandlerContext;Ljava/lang/Object;)V ProtocolNegotiators.java:937
    #2 io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(Ljava/lang/Object;)V AbstractChannelHandlerContext.java:398
    #3 io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(Lio/netty/channel/AbstractChannelHandlerContext;Ljava/lang/Object;)V AbstractChannelHandlerContext.java:376
    #4 io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(Ljava/lang/Object;)Lio/netty/channel/ChannelHandlerContext; AbstractChannelHandlerContext.java:368
    grpc#5 io.grpc.netty.ProtocolNegotiators$ProtocolNegotiationHandler.fireProtocolNegotiationEvent(Lio/netty/channel/ChannelHandlerContext;)V ProtocolNegotiators.java:1107
    grpc#6 io.grpc.netty.ProtocolNegotiators$WaitUntilActiveHandler.channelActive(Lio/netty/channel/ChannelHandlerContext;)V ProtocolNegotiators.java:1011
    ...

  Previous read of size 4 at 0x00008db31e2c by thread T4 (mutexes: write M0, write M1, write M2, write M3):
    #0 io.grpc.netty.NettyClientHandler.getAttributes()Lio/grpc/Attributes; NettyClientHandler.java:345
    #1 io.grpc.netty.NettyClientTransport.getAttributes()Lio/grpc/Attributes; NettyClientTransport.java:387
    #2 io.grpc.netty.NettyClientTransport.newStream(Lio/grpc/MethodDescriptor;Lio/grpc/Metadata;Lio/grpc/CallOptions;[Lio/grpc/ClientStreamTracer;)Lio/grpc/internal/ClientStream; NettyClientTransport.java:198
    #3 io.grpc.netty.NettyClientTransportTest$Rpc.<init>(Lio/grpc/netty/NettyClientTransport;Lio/grpc/Metadata;)V NettyClientTransportTest.java:953
    #4 io.grpc.netty.NettyClientTransportTest.huffmanCodingShouldNotBePerformed()V NettyClientTransportTest.java:631
    ...
```

```
  Read of size 4 at 0x00008f983a3c by thread T4 (mutexes: write M0, write M1):
    #0 io.grpc.netty.NettyClientHandler.getAttributes()Lio/grpc/Attributes; NettyClientHandler.java:345
    #1 io.grpc.netty.NettyClientTransport.getAttributes()Lio/grpc/Attributes; NettyClientTransport.java:387
    #2 io.grpc.netty.NettyClientTransport.newStream(Lio/grpc/MethodDescriptor;Lio/grpc/Metadata;Lio/grpc/CallOptions;[Lio/grpc/ClientStreamTracer;)Lio/grpc/internal/ClientStream; NettyClientTransport.java:198
    #3 io.grpc.netty.NettyClientTransportTest$Rpc.<init>(Lio/grpc/netty/NettyClientTransport;Lio/grpc/Metadata;)V NettyClientTransportTest.java:973
    #4 io.grpc.netty.NettyClientTransportTest$Rpc.<init>(Lio/grpc/netty/NettyClientTransport;)V NettyClientTransportTest.java:969
    grpc#5 io.grpc.netty.NettyClientTransportTest.handlerExceptionDuringNegotiatonPropagatesToStatus()V NettyClientTransportTest.java:425
    ...

  Previous write of size 4 at 0x00008f983a3c by thread T56:
    #0 io.grpc.netty.NettyClientHandler$FrameListener.onSettingsRead(Lio/netty/channel/ChannelHandlerContext;Lio/netty/handler/codec/http2/Http2Settings;)V NettyClientHandler.java:960
    ...
```

Author: ejona86

netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java

@@ -37,6 +37,8 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.google.common.base.Optional;
@@ -95,6 +97,7 @@
 import io.netty.handler.ssl.ClientAuth;
 import io.netty.handler.ssl.SslContext;
 import io.netty.util.AsciiString;
+import io.netty.util.ReferenceCountUtil;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -187,6 +190,7 @@ public void addDefaultUserAgent() throws Exception {
     startServer();
     NettyClientTransport transport = newTransport(newNegotiator());
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
 
     // Send a single RPC and wait for the response.
     new Rpc(transport).halfClose().waitForResponse();
@@ -244,6 +248,7 @@ public void overrideDefaultUserAgent() throws Exception {
     NettyClientTransport transport = newTransport(newNegotiator(),
         DEFAULT_MAX_MESSAGE_SIZE, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, "testUserAgent", true);
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
 
     new Rpc(transport, new Metadata()).halfClose().waitForResponse();
 
@@ -261,6 +266,7 @@ public void maxMessageSizeShouldBeEnforced() throws Throwable {
     NettyClientTransport transport = newTransport(newNegotiator(),
         1, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, null, true);
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
 
     try {
       // Send a single RPC and wait for the response.
@@ -287,6 +293,7 @@ public void creatingMultipleTlsTransportsShouldSucceed() throws Exception {
       NettyClientTransport transport = newTransport(negotiator);
       callMeMaybe(transport.start(clientTransportListener));
     }
+    verify(clientTransportListener, timeout(5000).times(2)).transportReady();
 
     // Send a single RPC on each transport.
     final List<Rpc> rpcs = new ArrayList<>(transports.size());
@@ -316,6 +323,7 @@ public void run() {
             failureStatus.asRuntimeException());
       }
     });
+    verify(clientTransportListener, timeout(5000)).transportTerminated();
 
     Rpc rpc = new Rpc(transport).halfClose();
     try {
@@ -349,6 +357,7 @@ public void tlsNegotiationFailurePropagatesToStatus() throws Exception {
     ProtocolNegotiator negotiator = ProtocolNegotiators.tls(clientContext);
     final NettyClientTransport transport = newTransport(negotiator);
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportTerminated();
 
     Rpc rpc = new Rpc(transport).halfClose();
     try {
@@ -378,6 +387,7 @@ public void channelExceptionDuringNegotiatonPropagatesToStatus() throws Exceptio
     callMeMaybe(transport.start(clientTransportListener));
     final Status failureStatus = Status.UNAVAILABLE.withDescription("oh noes!");
     transport.channel().pipeline().fireExceptionCaught(failureStatus.asRuntimeException());
+    verify(clientTransportListener, timeout(5000)).transportTerminated();
 
     Rpc rpc = new Rpc(transport).halfClose();
     try {
@@ -409,6 +419,7 @@ public void run() {
         }
       }
     });
+    verify(clientTransportListener, timeout(5000)).transportTerminated();
 
     Rpc rpc = new Rpc(transport).halfClose();
     try {
@@ -428,6 +439,7 @@ public void bufferedStreamsShouldBeClosedWhenConnectionTerminates() throws Excep
 
     NettyClientTransport transport = newTransport(newNegotiator());
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
 
     // Send a dummy RPC in order to ensure that the updated SETTINGS_MAX_CONCURRENT_STREAMS
     // has been received by the remote endpoint.
@@ -579,6 +591,7 @@ public void maxHeaderListSizeShouldBeEnforcedOnClient() throws Exception {
     NettyClientTransport transport =
         newTransport(newNegotiator(), DEFAULT_MAX_MESSAGE_SIZE, 1, null, true);
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
 
     try {
       // Send a single RPC and wait for the response.
@@ -612,6 +625,7 @@ public void huffmanCodingShouldNotBePerformed() throws Exception {
         longStringOfA);
 
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
 
     AtomicBoolean foundExpectedHeaderBytes = new AtomicBoolean(false);
 
@@ -641,6 +655,7 @@ public void maxHeaderListSizeShouldBeEnforcedOnServer() throws Exception {
 
     NettyClientTransport transport = newTransport(newNegotiator());
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
 
     try {
       // Send a single RPC and wait for the response.
@@ -685,6 +700,7 @@ public void clientStreamGetsAttributes() throws Exception {
     startServer();
     NettyClientTransport transport = newTransport(newNegotiator());
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
     Rpc rpc = new Rpc(transport).halfClose();
     rpc.waitForResponse();
 
@@ -703,6 +719,7 @@ public void keepAliveEnabled() throws Exception {
     NettyClientTransport transport = newTransport(newNegotiator(), DEFAULT_MAX_MESSAGE_SIZE,
         GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, null /* user agent */, true /* keep alive */);
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
     Rpc rpc = new Rpc(transport).halfClose();
     rpc.waitForResponse();
 
@@ -715,6 +732,7 @@ public void keepAliveDisabled() throws Exception {
     NettyClientTransport transport = newTransport(newNegotiator(), DEFAULT_MAX_MESSAGE_SIZE,
         GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, null /* user agent */, false /* keep alive */);
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
     Rpc rpc = new Rpc(transport).halfClose();
     rpc.waitForResponse();
 
@@ -808,6 +826,7 @@ public void tlsNegotiationServerExecutorShouldSucceed() throws Exception {
     assertEquals(true, clientExecutorPool.isInUse());
     final NettyClientTransport transport = newTransport(negotiator);
     callMeMaybe(transport.start(clientTransportListener));
+    verify(clientTransportListener, timeout(5000)).transportReady();
     Rpc rpc = new Rpc(transport).halfClose();
     rpc.waitForResponse();
     // closing the negotiators should return the executors back to pool, and release the resource
@@ -1098,9 +1117,15 @@ public NoopHandler(GrpcHttp2ConnectionHandler grpcHandler) {
       this.grpcHandler = grpcHandler;
     }
 
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+      // Prevent any data being passed to NettyClientHandler
+      ReferenceCountUtil.release(msg);
+    }
+
     @Override
     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
-      ctx.pipeline().addBefore(ctx.name(), null, grpcHandler);
+      ctx.pipeline().addAfter(ctx.name(), null, grpcHandler);
     }
 
     public void fail(ChannelHandlerContext ctx, Throwable cause) {