From 4044878c7197e33e567552d320ff5db195ca9b78 Mon Sep 17 00:00:00 2001
From: Adrien Grand <jpountz@gmail.com>
Date: Tue, 13 May 2014 11:04:23 +0200
Subject: [PATCH] Fix NPE when initializing an accepted socket in
 NettyTransport.

NettyTransport's ChannelPipelineFactory uses the instance variable
serverOpenChannels in order to create sockets. However, this instance variable
is set to null when stoping the netty transport, so if the transport tries to
stop and to initialize a socket at the same time you might hit the following
NullPointerException:

[2014-05-13 07:33:47,616][WARN ][netty.channel.socket.nio.AbstractNioSelector] Failed to initialize an accepted socket.
java.lang.NullPointerException: handler
	at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.<init>(DefaultChannelPipeline.java:725)
	at org.jboss.netty.channel.DefaultChannelPipeline.init(DefaultChannelPipeline.java:667)
	at org.jboss.netty.channel.DefaultChannelPipeline.addLast(DefaultChannelPipeline.java:96)
	at org.elasticsearch.transport.netty.NettyTransport$2.getPipeline(NettyTransport.java:327)
	at org.jboss.netty.channel.socket.nio.NioServerBoss.registerAcceptedChannel(NioServerBoss.java:134)
	at org.jboss.netty.channel.socket.nio.NioServerBoss.process(NioServerBoss.java:104)
	at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:318)
	at org.jboss.netty.channel.socket.nio.NioServerBoss.run(NioServerBoss.java:42)
	at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108)
	at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
	at java.lang.Thread.run(Thread.java:745)

This fix ensures that the ChannelPipelineFactory always uses the channels that
have been used upon start, even if a stop request is issued concurrently.

Close #6144
---
 .../org/elasticsearch/transport/netty/NettyTransport.java    | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java b/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java
index b689da9fc00ea..fafc4ad98217d 100644
--- a/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java
+++ b/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java
@@ -300,7 +300,8 @@ public ChannelPipeline getPipeline() throws Exception {
             return;
         }
 
-        serverOpenChannels = new OpenChannelsHandler(logger);
+        final OpenChannelsHandler openChannels = new OpenChannelsHandler(logger);
+        this.serverOpenChannels = openChannels;
         if (blockingServer) {
             serverBootstrap = new ServerBootstrap(new OioServerSocketChannelFactory(
                     Executors.newCachedThreadPool(daemonThreadFactory(settings, "transport_server_boss")),
@@ -316,7 +317,7 @@ public ChannelPipeline getPipeline() throws Exception {
             @Override
             public ChannelPipeline getPipeline() throws Exception {
                 ChannelPipeline pipeline = Channels.pipeline();
-                pipeline.addLast("openChannels", serverOpenChannels);
+                pipeline.addLast("openChannels", openChannels);
                 SizeHeaderFrameDecoder sizeHeader = new SizeHeaderFrameDecoder();
                 if (maxCumulationBufferCapacity != null) {
                     if (maxCumulationBufferCapacity.bytes() > Integer.MAX_VALUE) {