Repository: netty/netty
Issue: HTTP/2 codec may not always call Http2Connection$Listener.onStreamRemoved
Labels: defect

If `Http2Connection$Listener.onStreamAdded` is called it is not always the case that `Http2Connection$Listener.onStreamRemoved` will be called. Some use cases (including `InboundHttp2ToHttpAdapter`) depend upon the `onStreamRemoved` method to be called in order to clean up allocated buffers.

[Http2ConnectionHandler.channelInactive](https://github.com/netty/netty/blob/4.1/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java#L180) currently only iterates over active streams, and this does not account for streams which may still exist but are not active.


Issue comments:
@nmittler - FYI

related to https://github.com/netty/netty/issues/4780#issuecomment-179849253

@Scottmitch yikes! Are you working on a patch?

@nmittler - I was planning on it, but haven't got around to it yet....does this one peek your interest?

pr pending


Pull request review comments:
Why not just take a `ChannelPromise`?

Just curious ... not sure how Netty handles this case in general: what if `promise` is a void promise? If the caller tries to call `addListener` it will fail ... maybe we don't care, since they're the one that passed in the void promise?

Don't we have a light-weight promise aggregator that basically does this?

Would it make sense to just loop through and close them all and then optionally clear the map after the loop?

Much better :)

If the user passed in a void promise and tries to add a listener to it ... returning the same promise is also done in other places in Netty (e.g. `channel.write()`) so I don't think this should be unexpected.

It makes unit testing easier (given all the Mock objects that we have, creating a `ChannelPromise` would require more setup) and there is no reason why the code has to have a `ChannelPromise` if it can work with a `Promise<Void>`.

[PromiseNotifier](https://github.com/netty/netty/blob/4.1/common/src/main/java/io/netty/util/concurrent/PromiseNotifier.java) may work ... not sure if this is "light weight" as it creates an array.

Possibly but I would rather not invalidate any logic (which may result in a connection error) that may be going on during active stream iteration. Granted how we are currently using it is when the connection is being shut (or has already been shut down) down anyways, but I think it is easy enough to respect the active stream iteration so I just did this ... WDYT?

I will move this code to a utility class and submit a followup PR (or just use PromiseNotifier after I fix a bug in it :smile: )

I'm just concerned about the complexity introduced by not using `ChannelPromise`.  It a real setup, it will always be a `ChannelPromise` so it would seem that some of this code could go away. How much harder does this make your tests? I seem to recall mocking `ChannelPromise` without too much of an issue, no?

> I'm just concerned about the complexity introduced by not using ChannelPromise

What complexity are you referring to? The instanceof to support the void promise check?

Yeah, the if gets a bit simpler that's all. I don't feel that strongly.


Patch that resolved the issue:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
index 589546dbe1b..6dae971b4a5 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
@@ -15,28 +15,16 @@
 
 package io.netty.handler.codec.http2;
 
-import static io.netty.handler.codec.http2.Http2CodecUtil.CONNECTION_STREAM_ID;
-import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
-import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_WEIGHT;
-import static io.netty.handler.codec.http2.Http2CodecUtil.MIN_WEIGHT;
-import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
-import static io.netty.handler.codec.http2.Http2Error.REFUSED_STREAM;
-import static io.netty.handler.codec.http2.Http2Exception.closedStreamError;
-import static io.netty.handler.codec.http2.Http2Exception.connectionError;
-import static io.netty.handler.codec.http2.Http2Exception.streamError;
-import static io.netty.handler.codec.http2.Http2Stream.State.CLOSED;
-import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
-import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_REMOTE;
-import static io.netty.handler.codec.http2.Http2Stream.State.IDLE;
-import static io.netty.handler.codec.http2.Http2Stream.State.OPEN;
-import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_LOCAL;
-import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_REMOTE;
-import static io.netty.util.internal.ObjectUtil.checkNotNull;
 import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http2.Http2Stream.State;
 import io.netty.util.collection.IntCollections;
 import io.netty.util.collection.IntObjectHashMap;
 import io.netty.util.collection.IntObjectMap;
+import io.netty.util.collection.IntObjectMap.PrimitiveEntry;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
+import io.netty.util.concurrent.Promise;
 import io.netty.util.internal.EmptyArrays;
 import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.SystemPropertyUtil;
@@ -51,6 +39,25 @@
 import java.util.List;
 import java.util.Queue;
 import java.util.Set;
+
+import static io.netty.handler.codec.http2.Http2CodecUtil.CONNECTION_STREAM_ID;
+import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
+import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_WEIGHT;
+import static io.netty.handler.codec.http2.Http2CodecUtil.MIN_WEIGHT;
+import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
+import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
+import static io.netty.handler.codec.http2.Http2Error.REFUSED_STREAM;
+import static io.netty.handler.codec.http2.Http2Exception.closedStreamError;
+import static io.netty.handler.codec.http2.Http2Exception.connectionError;
+import static io.netty.handler.codec.http2.Http2Exception.streamError;
+import static io.netty.handler.codec.http2.Http2Stream.State.CLOSED;
+import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
+import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_REMOTE;
+import static io.netty.handler.codec.http2.Http2Stream.State.IDLE;
+import static io.netty.handler.codec.http2.Http2Stream.State.OPEN;
+import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_LOCAL;
+import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_REMOTE;
+import static io.netty.util.internal.ObjectUtil.checkNotNull;
 import static java.lang.Math.max;
 
 /**
@@ -80,6 +87,7 @@ public class DefaultHttp2Connection implements Http2Connection {
      */
     final List<Listener> listeners = new ArrayList<Listener>(4);
     final ActiveStreams activeStreams;
+    Promise<Void> closePromise;
 
     /**
      * Creates a new connection with the given settings.
@@ -96,6 +104,71 @@ public DefaultHttp2Connection(boolean server) {
         streamMap.put(connectionStream.id(), connectionStream);
     }
 
+    /**
+     * Determine if {@link #close(Promise)} has been called and no more streams are allowed to be created.
+     */
+    final boolean isClosed() {
+        return closePromise != null;
+    }
+
+    @Override
+    public Future<Void> close(final Promise<Void> promise) {
+        checkNotNull(promise, "promise");
+        // Since we allow this method to be called multiple times, we must make sure that all the promises are notified
+        // when all streams are removed and the close operation completes.
+        if (closePromise != null) {
+            if (closePromise == promise) {
+                // Do nothing
+            } else if ((promise instanceof ChannelPromise) && ((ChannelPromise) closePromise).isVoid()) {
+                closePromise = promise;
+            } else {
+                closePromise.addListener(new FutureListener<Void>() {
+                    @Override
+                    public void operationComplete(Future<Void> future) throws Exception {
+                        if (future.isSuccess()) {
+                            promise.trySuccess(null);
+                        } else if (future.isCancelled()) {
+                            promise.cancel(false);
+                        } else {
+                            promise.tryFailure(future.cause());
+                        }
+                    }
+                });
+            }
+        } else {
+            closePromise = promise;
+        }
+        if (isStreamMapEmpty()) {
+            promise.trySuccess(null);
+            return promise;
+        }
+        Iterator<PrimitiveEntry<Http2Stream>> itr = streamMap.entries().iterator();
+        // We must take care while iterating the streamMap as to not modify while iterating in case there are other code
+        // paths iterating over the active streams.
+        if (activeStreams.allowModifications()) {
+            while (itr.hasNext()) {
+                Http2Stream stream = itr.next().value();
+                if (stream.id() != CONNECTION_STREAM_ID) {
+                    // If modifications of the activeStream map is allowed, then a stream close operation will also
+                    // modify the streamMap. We must prevent concurrent modifications to the streamMap, so use the
+                    // iterator to remove the current stream.
+                    itr.remove();
+                    stream.close();
+                }
+            }
+        } else {
+            while (itr.hasNext()) {
+                Http2Stream stream = itr.next().value();
+                if (stream.id() != CONNECTION_STREAM_ID) {
+                    // We are not allowed to make modifications, so the close calls will be executed after this
+                    // iteration completes.
+                    stream.close();
+                }
+            }
+        }
+        return closePromise;
+    }
+
     @Override
     public void addListener(Listener listener) {
         listeners.add(listener);
@@ -208,6 +281,13 @@ public boolean visit(Http2Stream stream) {
         }
     }
 
+    /**
+     * Determine if {@link #streamMap} only contains the connection stream.
+     */
+    private boolean isStreamMapEmpty() {
+        return streamMap.size() == 1;
+    }
+
     /**
      * Closed streams may stay in the priority tree if they have dependents that are in prioritizable states.
      * When a stream is requested to be removed we can only actually remove that stream when there are no more
@@ -220,7 +300,6 @@ public boolean visit(Http2Stream stream) {
     void removeStream(DefaultStream stream) {
         // [1] Check if this stream can be removed because it has no prioritizable descendants.
         if (stream.parent().removeChild(stream)) {
-            // Remove it from the map and priority tree.
             streamMap.remove(stream.id());
 
             for (int i = 0; i < listeners.size(); i++) {
@@ -230,6 +309,10 @@ void removeStream(DefaultStream stream) {
                     logger.error("Caught RuntimeException from listener onStreamRemoved.", e);
                 }
             }
+
+            if (closePromise != null && isStreamMapEmpty()) {
+                closePromise.trySuccess(null);
+            }
         }
     }
 
@@ -604,17 +687,16 @@ final void takeChild(DefaultStream child, boolean exclusive, List<ParentChangedE
                 // path is updated with the correct child.prioritizableForTree() value. Note that the removal operation
                 // may not be successful and may return null. This is because when an exclusive dependency is processed
                 // the children are removed in a previous recursive call but the child's parent link is updated here.
-                if (oldParent != null && oldParent.children.remove(child.id()) != null) {
-                    if (!child.isDescendantOf(oldParent)) {
-                        oldParent.decrementPrioritizableForTree(child.prioritizableForTree());
-                        if (oldParent.prioritizableForTree() == 0) {
-                            // There are a few risks with immediately removing nodes from the priority tree:
-                            // 1. We are removing nodes while we are potentially shifting the tree. There are no
-                            // concrete cases known but is risky because it could invalidate the data structure.
-                            // 2. We are notifying listeners of the removal while the tree is in flux. Currently the
-                            // codec listeners make no assumptions about priority tree structure when being notified.
-                            removeStream(oldParent);
-                        }
+                if (oldParent != null && oldParent.children.remove(child.id()) != null &&
+                        !child.isDescendantOf(oldParent)) {
+                    oldParent.decrementPrioritizableForTree(child.prioritizableForTree());
+                    if (oldParent.prioritizableForTree() == 0) {
+                        // There are a few risks with immediately removing nodes from the priority tree:
+                        // 1. We are removing nodes while we are potentially shifting the tree. There are no
+                        // concrete cases known but is risky because it could invalidate the data structure.
+                        // 2. We are notifying listeners of the removal while the tree is in flux. Currently the
+                        // codec listeners make no assumptions about priority tree structure when being notified.
+                        removeStream(oldParent);
                     }
                 }
 
@@ -1040,6 +1122,10 @@ private void checkNewStreamAllowed(int streamId, State state) throws Http2Except
             if ((state.localSideOpen() || state.remoteSideOpen()) && !canOpenStream()) {
                 throw connectionError(REFUSED_STREAM, "Maximum active streams violated for this endpoint.");
             }
+            if (isClosed()) {
+                throw connectionError(INTERNAL_ERROR, "Attempted to create stream id %d after connection was closed",
+                                      streamId);
+            }
         }
 
         private boolean isLocal() {
@@ -1066,7 +1152,6 @@ interface Event {
      * active streams in order to prevent modification while iterating.
      */
     private final class ActiveStreams {
-
         private final List<Listener> listeners;
         private final Queue<Event> pendingEvents = new ArrayDeque<Event>(4);
         private final Set<Http2Stream> streams = new LinkedHashSet<Http2Stream>();
@@ -1157,7 +1242,7 @@ void removeFromActiveStreams(DefaultStream stream) {
             removeStream(stream);
         }
 
-        private boolean allowModifications() {
+        boolean allowModifications() {
             return pendingIterations == 0;
         }
     }
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Connection.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Connection.java
index 380ab329a74..419e216982c 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Connection.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Connection.java
@@ -16,6 +16,8 @@
 package io.netty.handler.codec.http2;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.Promise;
 
 /**
  * Manager for the state of an HTTP/2 connection with the remote end-point.
@@ -292,6 +294,16 @@ interface Endpoint<F extends Http2FlowController> {
     interface PropertyKey {
     }
 
+    /**
+     * Close this connection. No more new streams can be created after this point and
+     * all streams that exists (active or otherwise) will be closed and removed.
+     * <p>Note if iterating active streams via {@link #forEachActiveStream(Http2StreamVisitor)} and an exception is
+     * thrown it is necessary to call this method again to ensure the close completes.
+     * @param promise Will be completed when all streams have been removed, and listeners have been notified.
+     * @return A future that will be completed when all streams have been removed, and listeners have been notified.
+     */
+    Future<Void> close(Promise<Void> promise);
+
     /**
      * Creates a new key that is unique within this {@link Http2Connection}.
      */
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java
index e8f92c14b70..5c975af55a1 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java
@@ -175,18 +175,9 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
             encoder().close();
             decoder().close();
 
-            final Http2Connection connection = connection();
-            // Check if there are streams to avoid the overhead of creating the ChannelFuture.
-            if (connection.numActiveStreams() > 0) {
-                final ChannelFuture future = ctx.newSucceededFuture();
-                connection.forEachActiveStream(new Http2StreamVisitor() {
-                    @Override
-                    public boolean visit(Http2Stream stream) throws Http2Exception {
-                        closeStream(stream, future);
-                        return true;
-                    }
-                });
-            }
+            // We need to remove all streams (not just the active ones).
+            // See https://github.com/netty/netty/issues/4838.
+            connection().close(ctx.voidPromise());
         }
 
         /**


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionTest.java
index 2e68c39ff12..bb4ace388d9 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionTest.java
@@ -15,6 +15,33 @@
 
 package io.netty.handler.codec.http2;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.DefaultEventLoopGroup;
+import io.netty.handler.codec.http2.Http2Connection.Endpoint;
+import io.netty.handler.codec.http2.Http2Stream.State;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
+import io.netty.util.concurrent.Promise;
+import io.netty.util.internal.PlatformDependent;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
 import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
 import static io.netty.handler.codec.http2.Http2CodecUtil.MIN_WEIGHT;
 import static org.junit.Assert.assertEquals;
@@ -35,25 +62,6 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
-import io.netty.handler.codec.http2.Http2Connection.Endpoint;
-import io.netty.handler.codec.http2.Http2Stream.State;
-import io.netty.util.internal.PlatformDependent;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicReference;
-
 /**
  * Tests for {@link DefaultHttp2Connection}.
  */
@@ -63,6 +71,7 @@ public class DefaultHttp2ConnectionTest {
 
     private DefaultHttp2Connection server;
     private DefaultHttp2Connection client;
+    private static DefaultEventLoopGroup group;
 
     @Mock
     private Http2Connection.Listener clientListener;
@@ -70,6 +79,16 @@ public class DefaultHttp2ConnectionTest {
     @Mock
     private Http2Connection.Listener clientListener2;
 
+    @BeforeClass
+    public static void beforeClass() {
+        group = new DefaultEventLoopGroup(2);
+    }
+
+    @AfterClass
+    public static void afterClass() {
+        group.shutdownGracefully();
+    }
+
     @Before
     public void setup() {
         MockitoAnnotations.initMocks(this);
@@ -84,6 +103,110 @@ public void getStreamWithoutStreamShouldReturnNull() {
         assertNull(server.stream(100));
     }
 
+    @Test
+    public void removeAllStreamsWithEmptyStreams() throws InterruptedException {
+        testRemoveAllStreams();
+    }
+
+    @Test
+    public void removeAllStreamsWithJustOneLocalStream() throws InterruptedException, Http2Exception {
+        client.local().createStream(3, false);
+        testRemoveAllStreams();
+    }
+
+    @Test
+    public void removeAllStreamsWithJustOneRemoveStream() throws InterruptedException, Http2Exception {
+        client.remote().createStream(2, false);
+        testRemoveAllStreams();
+    }
+
+    @Test
+    public void removeAllStreamsWithManyActiveStreams() throws InterruptedException, Http2Exception {
+        Endpoint<Http2RemoteFlowController> remote = client.remote();
+        Endpoint<Http2LocalFlowController> local = client.local();
+        for (int c = 3, s = 2; c < 5000; c += 2, s += 2) {
+            local.createStream(c, false);
+            remote.createStream(s, false);
+        }
+        testRemoveAllStreams();
+    }
+
+    @Test
+    public void removeAllStreamsWithNonActiveStreams() throws InterruptedException, Http2Exception {
+        client.local().createIdleStream(3);
+        client.remote().createIdleStream(2);
+        testRemoveAllStreams();
+    }
+
+    @Test
+    public void removeAllStreamsWithNonActiveAndActiveStreams() throws InterruptedException, Http2Exception {
+        client.local().createIdleStream(3);
+        client.remote().createIdleStream(2);
+        client.local().createStream(5, false);
+        client.remote().createStream(4, true);
+        testRemoveAllStreams();
+    }
+
+    @Test
+    public void removeAllStreamsWhileIteratingActiveStreams() throws InterruptedException, Http2Exception {
+        final Endpoint<Http2RemoteFlowController> remote = client.remote();
+        final Endpoint<Http2LocalFlowController> local = client.local();
+        for (int c = 3, s = 2; c < 5000; c += 2, s += 2) {
+            local.createStream(c, false);
+            remote.createStream(s, false);
+        }
+        final Promise<Void> promise = group.next().newPromise();
+        final CountDownLatch latch = new CountDownLatch(client.numActiveStreams());
+        client.forEachActiveStream(new Http2StreamVisitor() {
+            @Override
+            public boolean visit(Http2Stream stream) throws Http2Exception {
+                client.close(promise).addListener(new FutureListener<Void>() {
+                    @Override
+                    public void operationComplete(Future<Void> future) throws Exception {
+                        assertTrue(promise.isDone());
+                        latch.countDown();
+                    }
+                });
+                return true;
+            }
+        });
+        assertTrue(latch.await(2, TimeUnit.SECONDS));
+    }
+
+    @Test
+    public void removeAllStreamsWhileIteratingActiveStreamsAndExceptionOccurs()
+            throws InterruptedException, Http2Exception {
+        final Endpoint<Http2RemoteFlowController> remote = client.remote();
+        final Endpoint<Http2LocalFlowController> local = client.local();
+        for (int c = 3, s = 2; c < 5000; c += 2, s += 2) {
+            local.createStream(c, false);
+            remote.createStream(s, false);
+        }
+        final Promise<Void> promise = group.next().newPromise();
+        final CountDownLatch latch = new CountDownLatch(1);
+        try {
+            client.forEachActiveStream(new Http2StreamVisitor() {
+                @Override
+                public boolean visit(Http2Stream stream) throws Http2Exception {
+                    // This close call is basically a noop, because the following statement will throw an exception.
+                    client.close(promise);
+                    // Do an invalid operation while iterating.
+                    remote.createStream(3, false);
+                    return true;
+                }
+            });
+        } catch (Http2Exception ignored) {
+            client.close(promise).addListener(new FutureListener<Void>() {
+                @Override
+                public void operationComplete(Future<Void> future) throws Exception {
+                    assertTrue(promise.isDone());
+                    latch.countDown();
+                }
+            });
+        }
+        assertTrue(latch.await(2, TimeUnit.SECONDS));
+    }
+
     @Test
     public void goAwayReceivedShouldCloseStreamsGreaterThanLastStream() throws Exception {
         Http2Stream stream1 = client.local().createStream(3, false);
@@ -1107,6 +1230,19 @@ public void listenerThrowShouldNotPreventOtherListenersFromBeingNotified() throw
         }
     }
 
+    private void testRemoveAllStreams() throws InterruptedException {
+        final CountDownLatch latch = new CountDownLatch(1);
+        final Promise<Void> promise = group.next().newPromise();
+        client.close(promise).addListener(new FutureListener<Void>() {
+            @Override
+            public void operationComplete(Future<Void> future) throws Exception {
+                assertTrue(promise.isDone());
+                latch.countDown();
+            }
+        });
+        assertTrue(latch.await(2, TimeUnit.SECONDS));
+    }
+
     private void incrementAndGetStreamShouldRespectOverflow(Endpoint<?> endpoint, int streamId) throws Http2Exception {
         assertTrue(streamId > 0);
         try {
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionHandlerTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionHandlerTest.java
index c17cc4d8819..ed9d1c06ea4 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionHandlerTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionHandlerTest.java
@@ -26,6 +26,7 @@
 import io.netty.channel.DefaultChannelPromise;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.GenericFutureListener;
+import io.netty.util.concurrent.Promise;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -269,11 +270,12 @@ public void verifyChannelHandlerCanBeReusedInPipeline() throws Exception {
         verify(decoder, atLeastOnce()).decodeFrame(eq(ctx), any(ByteBuf.class), Matchers.<List<Object>>any());
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void channelInactiveShouldCloseStreams() throws Exception {
         handler = newHandler();
         handler.channelInactive(ctx);
-        verify(stream).close();
+        verify(connection).close(any(Promise.class));
     }
 
     @Test