Repository: relayrides/pushy
Pull request #714 (2019-08-26 13:30:47)

close pending create channel futures when close apns channel pool
Hi jchambers, 
Thanks for your work on this great lib again. I got an issue today and liked to share it with you.

Environment:
jdk: 1.8.0_192
pushy: 0.13.9
netty: 4.1.37.Final
tcnative: netty-tcnative-boringssl-static-2.0.25.Final-linux-x86_64

The exception stack was like this:
```
[WARN] [nioEventLoopGroup-3-1] 08-26 17:39:56,053 [ChannelInitializer] - Failed to initialize a channel. Closing: [id: 0x4de06504]                                                                                                      
java.lang.NullPointerException: ctx                                                                                                                                                                                                             at io.netty.internal.tcnative.SSL.newSSL(Native Method) ~[netty-tcnative-boringssl-static-2.0.25.Final-linux-x86_64.jar:2.0.25.Final]                                                                                                   at io.netty.handler.ssl.ReferenceCountedOpenSslEngine.<init>(ReferenceCountedOpenSslEngine.java:324) ~[netty-handler-4.1.37.Final.jar:4.1.37.Final]                                                                             
        at io.netty.handler.ssl.ReferenceCountedOpenSslContext.newEngine0(ReferenceCountedOpenSslContext.java:423) ~[netty-handler-4.1.37.Final.jar:4.1.37.Final]                                                                       
        at io.netty.handler.ssl.ReferenceCountedOpenSslContext.newHandler(ReferenceCountedOpenSslContext.java:403) ~[netty-handler-4.1.37.Final.jar:4.1.37.Final]                                                                       
        at io.netty.handler.ssl.SslContext.newHandler(SslContext.java:935) ~[netty-handler-4.1.37.Final.jar:4.1.37.Final]                                                                                                                       at com.turo.pushy.apns.ApnsChannelFactory$1.initChannel(ApnsChannelFactory.java:124) ~[pushy-0.13.9.jar:?]                                                                                                                              at com.turo.pushy.apns.ApnsChannelFactory$1.initChannel(ApnsChannelFactory.java:114) ~[pushy-0.13.9.jar:?]                                                                                                                      
        at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:129) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]                                                                                                
        at io.netty.channel.ChannelInitializer.handlerAdded(ChannelInitializer.java:112) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]                                                                                               
        at io.netty.channel.AbstractChannelHandlerContext.callHandlerAdded(AbstractChannelHandlerContext.java:964) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]                                                                     
        at io.netty.channel.DefaultChannelPipeline.callHandlerAdded0(DefaultChannelPipeline.java:610) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.channel.DefaultChannelPipeline.access$100(DefaultChannelPipeline.java:46) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.channel.DefaultChannelPipeline$PendingHandlerAddedTask.execute(DefaultChannelPipeline.java:1461) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.channel.DefaultChannelPipeline.callHandlerAddedForAllHandlers(DefaultChannelPipeline.java:1126) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.channel.DefaultChannelPipeline.invokeHandlerAddedIfNeeded(DefaultChannelPipeline.java:651) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.channel.AbstractChannel$AbstractUnsafe.register0(AbstractChannel.java:503) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.channel.AbstractChannel$AbstractUnsafe.access$200(AbstractChannel.java:416) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.channel.AbstractChannel$AbstractUnsafe$1.run(AbstractChannel.java:475) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) ~[netty-common-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416) ~[netty-common-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515) ~[netty-transport-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) ~[netty-common-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.37.Final.jar:4.1.37.Final]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.37.Final.jar:4.1.37.Final]
        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_192]
```

When [the ApnsChannelPool is closing](https://github.com/relayrides/pushy/blob/pushy-0.13.9/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java#L256)  there are still some [pending create channel future waiting for fulfill](https://github.com/relayrides/pushy/blob/pushy-0.13.9/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java#L63) . After [the channelFactory was closed](https://github.com/relayrides/pushy/blob/pushy-0.13.9/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java#L263), [the sslContext in ApnsChannelFactory was released](https://github.com/relayrides/pushy/blob/pushy-0.13.9/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelFactory.java#L261), while maybe there's still [a create channel future trying to create a new channel with the released sslContext](https://github.com/relayrides/pushy/blob/pushy-0.13.9/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelFactory.java#L209), so the exception like above was thrown.

I think maybe we can cancel all the `pendingCreateChannelFutures` in [ApnsChannelPool.close](https://github.com/relayrides/pushy/blob/pushy-0.13.9/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java#L256) before close the `ApnsChannelFactory`. What do you think?

jchambers: Yikes—I totally missed the notification when this came in. I'll review it shortly. Thank you for the contribution, and please accept my apologies for the delay!
ylgrgyq: Yes, of course.
Done. Please have a look.
jchambers: It looks like everything did get squashed down to a single commit (thank you!), but isn't current with `master`. Would you mind giving it another shot?
ylgrgyq: Ah, sorry I forgot to fetch master branch from upstream and just rebased on my local master branch.
I think it's OK now. Please have a look again. Thanks. 
jchambers: > I forgot to fetch master branch from upstream and just rebased on my local master branch.

Yeah—forgetting to fetch is pretty much my full-time job ;)

Thanks very much!

Patch:
diff --git a/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java b/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java
index 53f6d73e..c493e25a 100644
--- a/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java
+++ b/pushy/src/main/java/com/turo/pushy/apns/ApnsChannelPool.java
@@ -254,19 +254,37 @@ class ApnsChannelPool {
      * @return a {@code Future} that will be completed when all resources held by this pool have been released
      */
     public Future<Void> close() {
-        return this.allChannels.close().addListener(new GenericFutureListener<Future<Void>>() {
+        final Promise<Void> closeDonePromise = new DefaultPromise<>(this.executor);
+        this.allChannels.close().addListener(new GenericFutureListener<Future<Void>>() {
             @Override
-            public void operationComplete(final Future<Void> future) throws Exception {
+            public void operationComplete(final Future<Void> future) {
                 ApnsChannelPool.this.isClosed = true;
 
-                if (ApnsChannelPool.this.channelFactory instanceof Closeable) {
-                    ((Closeable) ApnsChannelPool.this.channelFactory).close();
-                }
+                final Promise<Void> waitForPendingCreateChannelFuturesPromise = new DefaultPromise<>(ApnsChannelPool.this.executor);
+                final PromiseCombiner combiner = new PromiseCombiner(ApnsChannelPool.this.executor);
 
-                for (final Promise<Channel> acquisitionPromise : ApnsChannelPool.this.pendingAcquisitionPromises) {
-                    acquisitionPromise.tryFailure(POOL_CLOSED_EXCEPTION);
+                for (final Future<Channel> f : pendingCreateChannelFutures) {
+                    combiner.add(f);
                 }
+
+                combiner.finish(waitForPendingCreateChannelFuturesPromise);
+
+                waitForPendingCreateChannelFuturesPromise.addListener(new GenericFutureListener<Future<? super Void>>() {
+                    @Override
+                    public void operationComplete(Future<? super Void> future) throws Exception {
+                        if (ApnsChannelPool.this.channelFactory instanceof Closeable) {
+                            ((Closeable) ApnsChannelPool.this.channelFactory).close();
+                        }
+
+                        for (final Promise<Channel> acquisitionPromise : ApnsChannelPool.this.pendingAcquisitionPromises) {
+                            acquisitionPromise.tryFailure(POOL_CLOSED_EXCEPTION);
+                        }
+
+                        closeDonePromise.setSuccess(null);
+                    }
+                });
             }
         });
+        return closeDonePromise;
     }
 }


Test patch:
diff --git a/pushy/src/test/java/com/turo/pushy/apns/ApnsChannelPoolTest.java b/pushy/src/test/java/com/turo/pushy/apns/ApnsChannelPoolTest.java
index 395c380c..f1d3651a 100644
--- a/pushy/src/test/java/com/turo/pushy/apns/ApnsChannelPoolTest.java
+++ b/pushy/src/test/java/com/turo/pushy/apns/ApnsChannelPoolTest.java
@@ -214,4 +214,34 @@ public class ApnsChannelPoolTest {
         assertTrue(pendingFuture.await(1000));
         assertFalse(pendingFuture.isSuccess());
     }
+
+    @Test
+    public void testPendingCreateChannelFutureDuringPoolClosure() throws Exception {
+        final Promise<Void> createSuccess = new DefaultPromise<>(EVENT_EXECUTOR);
+        final PooledObjectFactory<Channel> factory = new PooledObjectFactory<Channel>() {
+            @Override
+            public Future<Channel> create(final Promise<Channel> promise) {
+                EVENT_EXECUTOR.schedule(new Runnable() {
+                    @Override
+                    public void run() {
+                        promise.trySuccess(new TestChannel(true));
+                        createSuccess.setSuccess(null);
+                    }
+                }, 1, TimeUnit.SECONDS);
+                return promise;
+            }
+
+            @Override
+            public Future<Void> destroy(final Channel channel, final Promise<Void> promise) {
+                promise.trySuccess(null);
+                return promise;
+            }
+        };
+
+        final ApnsChannelPool pool = new ApnsChannelPool(factory, 1, EVENT_EXECUTOR, this.metricsListener);
+
+        pool.acquire();
+        pool.close().await();
+        assertTrue(createSuccess.isSuccess());
+    }
 }