AbstractBootstrap can crash instead of failing promise, close #5387

Motivation:
When `ChannelFactory#newChannel` crashed, `AbstractBootstrap#initAndRegister` propagates the exception to the caller instead of failing the promise.

Modifications:
- Catch exceptions from `ChannelFactory#newChannel`.
- Notify promise of such failure.

Result:
`AbstractBootstrap` gracefully handles connect failures.

transport/src/main/java/io/netty/bootstrap/AbstractBootstrap.java

@@ -314,11 +314,15 @@ public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C ext
     }
 
     final ChannelFuture initAndRegister() {
-        final Channel channel = channelFactory.newChannel();
+        Channel channel = null;
         try {
+            channel = channelFactory.newChannel();
             init(channel);
         } catch (Throwable t) {
-            channel.unsafe().closeForcibly();
+            if (channel != null) {
+                // channel can be null if newChannel crashed (eg SocketException("too many open files"))
+                channel.unsafe().closeForcibly();
+            }
             // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
             return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
         }

transport/src/test/java/io/netty/bootstrap/BootstrapTest.java

@@ -69,7 +69,6 @@ public class BootstrapTest {
 
     @Test(timeout = 10000)
     public void testBindDeadLock() throws Exception {
-
         final Bootstrap bootstrapA = new Bootstrap();
         bootstrapA.group(groupA);
         bootstrapA.channel(LocalChannel.class);
@@ -106,7 +105,6 @@ public class BootstrapTest {
 
     @Test(timeout = 10000)
     public void testConnectDeadLock() throws Exception {
-
         final Bootstrap bootstrapA = new Bootstrap();
         bootstrapA.group(groupA);
         bootstrapA.channel(LocalChannel.class);
@@ -234,7 +232,6 @@ public class BootstrapTest {
 
     @Test
     public void testAsyncResolutionSuccess() throws Exception {
-
         final Bootstrap bootstrapA = new Bootstrap();
         bootstrapA.group(groupA);
         bootstrapA.channel(LocalChannel.class);
@@ -253,7 +250,6 @@ public class BootstrapTest {
 
     @Test
     public void testAsyncResolutionFailure() throws Exception {
-
         final Bootstrap bootstrapA = new Bootstrap();
         bootstrapA.group(groupA);
         bootstrapA.channel(LocalChannel.class);
@@ -275,6 +271,28 @@ public class BootstrapTest {
         assertThat(connectFuture.channel().isOpen(), is(false));
     }
 
+    @Test
+    public void testChannelFactoryFailureNotifiesPromise() throws Exception {
+        final RuntimeException exception = new RuntimeException("newChannel crash");
+
+        final Bootstrap bootstrap = new Bootstrap()
+                .handler(dummyHandler)
+                .group(groupA)
+                .channelFactory(new ChannelFactory<Channel>() {
+            @Override
+            public Channel newChannel() {
+                throw exception;
+            }
+        });
+
+        ChannelFuture connectFuture = bootstrap.connect(LocalAddress.ANY);
+
+        // Should fail with the RuntimeException.
+        assertThat(connectFuture.await(10000), is(true));
+        assertThat(connectFuture.cause(), sameInstance((Throwable) exception));
+        assertThat(connectFuture.channel(), is(nullValue()));
+    }
+
     private static final class DelayedEventLoopGroup extends DefaultEventLoop {
         @Override
         public ChannelFuture register(final Channel channel, final ChannelPromise promise) {