netty: Fix client keepalive initialization (again)

d116cc9 fixed the NPE, but the initialization of the manager happened
_after_ newHandler() was called, so a null manager was passed to the
handler.

Fixes #2828

Author: ejona86

netty/src/main/java/io/grpc/netty/NettyClientTransport.java

@@ -55,6 +55,7 @@
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http2.StreamBufferingEncoder.Http2ChannelClosedException;
@@ -165,14 +166,21 @@ public ClientStream newStream(MethodDescriptor<?, ?> method, Metadata headers) {
   public Runnable start(Listener transportListener) {
     lifecycleManager = new ClientTransportLifecycleManager(
         Preconditions.checkNotNull(transportListener, "listener"));
+    EventLoop eventLoop = group.next();
+    if (enableKeepAlive) {
+      keepAliveManager = new KeepAliveManager(
+          new ClientKeepAlivePinger(this), eventLoop, keepAliveDelayNanos, keepAliveTimeoutNanos,
+          false);
+    }
 
-    handler = newHandler();
+    handler = NettyClientHandler.newHandler(lifecycleManager, keepAliveManager, flowControlWindow,
+        maxHeaderListSize, Ticker.systemTicker());
     HandlerSettings.setAutoWindow(handler);
 
     negotiationHandler = negotiator.newHandler(handler);
 
     Bootstrap b = new Bootstrap();
-    b.group(group);
+    b.group(eventLoop);
     b.channel(channelType);
     if (NioSocketChannel.class.isAssignableFrom(channelType)) {
       b.option(SO_KEEPALIVE, true);
@@ -231,10 +239,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
       }
     });
 
-    if (enableKeepAlive) {
-      keepAliveManager = new KeepAliveManager(
-          new ClientKeepAlivePinger(this), channel.eventLoop(), keepAliveDelayNanos,
-          keepAliveTimeoutNanos, false);
+    if (keepAliveManager != null) {
       keepAliveManager.onTransportStarted();
     }
 
@@ -307,9 +312,4 @@ private Status statusFromFailedFuture(ChannelFuture f) {
     }
     return Utils.statusFromThrowable(t);
   }
-
-  private NettyClientHandler newHandler() {
-    return NettyClientHandler.newHandler(lifecycleManager, keepAliveManager, flowControlWindow,
-        maxHeaderListSize, Ticker.systemTicker());
-  }
 }