Repository: netty/netty
Issue: Investigate exception handling WRT Http2Connection.Listener notifications
Labels: defect

We should take a closer look at how we notify the listeners of connection events and consider the proper course of action if a listener callback throws.


Issue comments:
+1. Thanks for reporting!

@nmittler - I'll take this one.  As part of the resolution to https://github.com/netty/netty/issues/3550 I needed to do additional iteration over methods which use listeners and so I had to do something about this issue.  My approach was to catch `RuntimeExceptions` and only throw the last one:

``` java
RuntimeException lastException = null;
for (Listener l : listeneres) {
  try {
    l.notify(...);
   } catch (RuntimeException e) {
    lastException = e;
   }
}
if (lastException != null) {
  throw lastException;
}
```

Does this make sense?  The goal being 1 listener's exception shouldn't prevent other listener's from being notified.  However I am throwing 1 exception at the end of notification...just to ensure the exception is not swallowed...WDYT?

@Scottmitch yep that sgtm ... that's sort of what I was thinking as well.  Make it so!

@nmittler - Fair warning...it is ugly to ensure we process all events and complete all tree operations and also handle the `RuntimeExceptions` :(

@nmittler - An alternative approach could be that we document the `Http2Connection.Listener` interface such that thrown exceptions will result in the connection being closed (`GO_AWAY internal error`).  This way we don't have to worry about tracking exceptions in recursive calls, and making sure state (stream state, priority tree, etc...) is maintained in the presence of exceptions...WDYT?

@Scottmitch yeah, that actually sounds better to me.  Let's keep it simple and just update the docs.

@nmittler - I have to investigate but there may be code changes to ensure this always results in a GO_AWAY.  For example the removal policy stuff makes this a bit more complicated.  Perhaps we should handle this issue after https://github.com/netty/netty/issues/3448?

@Scottmitch yeah that's fine.

Reminder to me...also fix https://github.com/netty/netty/pull/3577/files#r27757163.

@nmittler - I have looked at this a bit more.  When events occur as a result of read operations I think we ultimately catch the `RuntimeExceptions` and translate these to GO_AWAY frames.  However on the send side we don't do this, but instead we sometimes (depending on if try/catch blocks are placed appropriately) fail the promise that was provided in the `writeXXX` call.  My concern is that we are in a situation where we use listeners to update stream state, and it is possible that 1 listener (potentially user listener) could interrupt this process, and then we just continue processing events while in a bad state.  Should we translate runtime exceptions on the write side into `lifecycleManager` failure events?  Any other ideas?

@Scottmitch yeah, I think that makes sense.  So fail the promise and send RST_STREAM/GO_AWAY.  Are you thinking of just adding this as try/catch logic to each encoder.writeXXX method?

@nmittler - I will investigate more but I think so....it is not ideal to handle these events up the call stack because then the classes which invoke the listeners are implicitly depending on those higher level classes to be present to clean up....I think the other options are to track exceptions at the call sites and either log them or throw the last one...but as I said this gets ugly...especially if we want to propagate 1 of the exceptions in the recursive calls we have.

@nmittler - A "cleaner" alternative (from us tracking the exceptions perspective) is just to "log and swallow" these types of exceptions at the call site.  We won't have to worry about tracking/propagating the exceptions...but swallowing exceptions is terrible too....

@Scottmich actually, I think I like logging (WARN?) better. Do you want to throw a PR together?

@nmittler - I'm glad you said that because I just finished a PR for this..and just added a unit test for it ;)

@nmittler - I updated the javadocs to describe this behavior (and logged it as an error).

#3615 cherry-picked


Pull request review comments:
Just noticed this dependency.  @nmittler - Is there no other way? :)

Nooooo! Let's just use `AtomicReference` :-)

+1 for `AtomicReference` :)

Done :)

Fixed in this PR for simplicity


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 6b9b7059832..a83436b6493 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
@@ -40,6 +40,8 @@
 import io.netty.util.collection.IntObjectMap;
 import io.netty.util.collection.PrimitiveCollections;
 import io.netty.util.internal.PlatformDependent;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
 
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -55,6 +57,7 @@
  * Simple implementation of {@link Http2Connection}.
  */
 public class DefaultHttp2Connection implements Http2Connection {
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultHttp2Connection.class);
     // Fields accessed by inner classes
     final IntObjectMap<Http2Stream> streamMap = new IntObjectHashMap<Http2Stream>();
     final ConnectionStream connectionStream = new ConnectionStream();
@@ -164,8 +167,12 @@ public boolean goAwayReceived() {
     @Override
     public void goAwayReceived(final int lastKnownStream, long errorCode, ByteBuf debugData) {
         localEndpoint.lastKnownStream(lastKnownStream);
-        for (Listener listener : listeners) {
-            listener.onGoAwayReceived(lastKnownStream, errorCode, debugData);
+        for (int i = 0; i < listeners.size(); ++i) {
+            try {
+                listeners.get(i).onGoAwayReceived(lastKnownStream, errorCode, debugData);
+            } catch (RuntimeException e) {
+                logger.error("Caught RuntimeException from listener onGoAwayReceived.", e);
+            }
         }
 
         try {
@@ -191,8 +198,12 @@ public boolean goAwaySent() {
     @Override
     public void goAwaySent(final int lastKnownStream, long errorCode, ByteBuf debugData) {
         remoteEndpoint.lastKnownStream(lastKnownStream);
-        for (Listener listener : listeners) {
-            listener.onGoAwaySent(lastKnownStream, errorCode, debugData);
+        for (int i = 0; i < listeners.size(); ++i) {
+            try {
+                listeners.get(i).onGoAwaySent(lastKnownStream, errorCode, debugData);
+            } catch (RuntimeException e) {
+                logger.error("Caught RuntimeException from listener onGoAwaySent.", e);
+            }
         }
 
         try {
@@ -226,7 +237,11 @@ void removeStream(DefaultStream stream) {
             streamMap.remove(stream.id());
 
             for (int i = 0; i < listeners.size(); i++) {
-                listeners.get(i).onStreamRemoved(stream);
+                try {
+                    listeners.get(i).onStreamRemoved(stream);
+                } catch (RuntimeException e) {
+                    logger.error("Caught RuntimeException from listener onStreamRemoved.", e);
+                }
             }
         }
     }
@@ -498,7 +513,11 @@ private boolean isPrioritizable() {
 
         private void notifyHalfClosed(Http2Stream stream) {
             for (int i = 0; i < listeners.size(); i++) {
-                listeners.get(i).onStreamHalfClosed(stream);
+                try {
+                    listeners.get(i).onStreamHalfClosed(stream);
+                } catch (RuntimeException e) {
+                    logger.error("Caught RuntimeException from listener onStreamHalfClosed.", e);
+                }
             }
         }
 
@@ -535,7 +554,11 @@ final void weight(short weight) {
                 final short oldWeight = this.weight;
                 this.weight = weight;
                 for (int i = 0; i < listeners.size(); i++) {
-                    listeners.get(i).onWeightChanged(this, oldWeight);
+                    try {
+                        listeners.get(i).onWeightChanged(this, oldWeight);
+                    } catch (RuntimeException e) {
+                        logger.error("Caught RuntimeException from listener onWeightChanged.", e);
+                    }
                 }
             }
         }
@@ -703,7 +726,11 @@ private static final class ParentChangedEvent {
          * @param l The listener to notify
          */
         public void notifyListener(Listener l) {
-            l.onPriorityTreeParentChanged(stream, oldParent);
+            try {
+                l.onPriorityTreeParentChanged(stream, oldParent);
+            } catch (RuntimeException e) {
+                logger.error("Caught RuntimeException from listener onPriorityTreeParentChanged.", e);
+            }
         }
     }
 
@@ -722,7 +749,11 @@ private void notifyParentChanged(List<ParentChangedEvent> events) {
 
     private void notifyParentChanging(Http2Stream stream, Http2Stream newParent) {
         for (int i = 0; i < listeners.size(); i++) {
-            listeners.get(i).onPriorityTreeParentChanging(stream, newParent);
+            try {
+                listeners.get(i).onPriorityTreeParentChanging(stream, newParent);
+            } catch (RuntimeException e) {
+                logger.error("Caught RuntimeException from listener onPriorityTreeParentChanging.", e);
+            }
         }
     }
 
@@ -869,7 +900,11 @@ private void addStream(DefaultStream stream) {
 
             // Notify the listeners of the event.
             for (int i = 0; i < listeners.size(); i++) {
-                listeners.get(i).onStreamAdded(stream);
+                try {
+                    listeners.get(i).onStreamAdded(stream);
+                } catch (RuntimeException e) {
+                    logger.error("Caught RuntimeException from listener onStreamAdded.", e);
+                }
             }
 
             notifyParentChanged(events);
@@ -971,6 +1006,9 @@ private static final class ActiveStreams {
         interface Event {
             /**
              * Trigger the original intention of this event. Expect to modify {@link #streams}.
+             * <p>
+             * If a {@link RuntimeException} object is thrown it will be logged and <strong>not propagated</strong>.
+             * Throwing from this method is not supported and is considered a programming error.
              */
             void process();
         }
@@ -1033,7 +1071,11 @@ public Http2Stream forEachActiveStream(Http2StreamVisitor visitor) throws Http2E
                         if (event == null) {
                             break;
                         }
-                        event.process();
+                        try {
+                            event.process();
+                        } catch (RuntimeException e) {
+                            logger.error("Caught RuntimeException while processing pending ActiveStreams$Event.", e);
+                        }
                     }
                 }
             }
@@ -1045,7 +1087,11 @@ void addToActiveStreams(DefaultStream stream) {
                 stream.createdBy().numActiveStreams++;
 
                 for (int i = 0; i < listeners.size(); i++) {
-                    listeners.get(i).onStreamActive(stream);
+                    try {
+                        listeners.get(i).onStreamActive(stream);
+                    } catch (RuntimeException e) {
+                        logger.error("Caught RuntimeException from listener onStreamActive.", e);
+                    }
                 }
             }
         }
@@ -1057,7 +1103,11 @@ void removeFromActiveStreams(DefaultStream stream) {
                     stream.createdBy().numActiveStreams--;
 
                     for (int i = 0; i < listeners.size(); i++) {
-                        listeners.get(i).onStreamClosed(stream);
+                        try {
+                            listeners.get(i).onStreamClosed(stream);
+                        } catch (RuntimeException e) {
+                            logger.error("Caught RuntimeException from listener onStreamClosed.", e);
+                        }
                     }
                 } finally {
                     // Mark this stream for removal.
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 80bd87379ec..07d6cf6fd6a 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
@@ -21,7 +21,6 @@
  * Manager for the state of an HTTP/2 connection with the remote end-point.
  */
 public interface Http2Connection {
-
     /**
      * Listener for life-cycle events for streams in this connection.
      */
@@ -29,23 +28,35 @@ interface Listener {
         /**
          * Notifies the listener that the given stream was added to the connection. This stream may
          * not yet be active (i.e. {@code OPEN} or {@code HALF CLOSED}).
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          */
         void onStreamAdded(Http2Stream stream);
 
         /**
          * Notifies the listener that the given stream was made active (i.e. {@code OPEN} or {@code HALF CLOSED}).
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          */
         void onStreamActive(Http2Stream stream);
 
         /**
          * Notifies the listener that the given stream is now {@code HALF CLOSED}. The stream can be
          * inspected to determine which side is {@code CLOSED}.
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          */
         void onStreamHalfClosed(Http2Stream stream);
 
         /**
          * Notifies the listener that the given stream is now {@code CLOSED} in both directions and will no longer
          * be accessible via {@link #forEachActiveStream(Http2StreamVisitor)}.
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          */
         void onStreamClosed(Http2Stream stream);
 
@@ -53,6 +64,9 @@ interface Listener {
          * Notifies the listener that the given stream has now been removed from the connection and
          * will no longer be returned via {@link Http2Connection#stream(int)}. The connection may
          * maintain inactive streams for some time before removing them.
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          */
         void onStreamRemoved(Http2Stream stream);
 
@@ -61,6 +75,9 @@ interface Listener {
          * in a top down order relative to the priority tree. This method will also be invoked after all tree
          * structure changes have been made and the tree is in steady state relative to the priority change
          * which caused the tree structure to change.
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          * @param stream The stream which had a parent change (new parent and children will be steady state)
          * @param oldParent The old parent which {@code stream} used to be a child of (may be {@code null})
          */
@@ -70,13 +87,19 @@ interface Listener {
          * Notifies the listener that a parent dependency is about to change
          * This is called while the tree is being restructured and so the tree
          * structure is not necessarily steady state.
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          * @param stream The stream which the parent is about to change to {@code newParent}
          * @param newParent The stream which will be the parent of {@code stream}
          */
         void onPriorityTreeParentChanging(Http2Stream stream, Http2Stream newParent);
 
         /**
-         * Notifies the listener that the weight has changed for {@code stream}
+         * Notifies the listener that the weight has changed for {@code stream}.
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          * @param stream The stream which the weight has changed
          * @param oldWeight The old weight for {@code stream}
          */
@@ -84,7 +107,9 @@ interface Listener {
 
         /**
          * Called when a {@code GOAWAY} frame was sent for the connection.
-         *
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          * @param lastStreamId the last known stream of the remote endpoint.
          * @param errorCode    the error code, if abnormal closure.
          * @param debugData    application-defined debug data.
@@ -97,7 +122,9 @@ interface Listener {
          * but is added here in order to simplify application logic for handling {@code GOAWAY} in a uniform way. An
          * application should generally not handle both events, but if it does this method is called second, after
          * notifying the {@link Http2FrameListener}.
-         *
+         * <p>
+         * If a {@link RuntimeException} is thrown it will be logged and <strong>not propagated</strong>.
+         * Throwing from this method is not supported and is considered a programming error.
          * @param lastStreamId the last known stream of the remote endpoint.
          * @param errorCode    the error code, if abnormal closure.
          * @param debugData    application-defined debug data.
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 6ea08cd72b3..51e63e60420 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
@@ -53,11 +53,11 @@
  */
 public class Http2ConnectionHandler extends ByteToMessageDecoder implements Http2LifecycleManager,
         ChannelOutboundHandler {
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(Http2ConnectionHandler.class);
     private final Http2ConnectionDecoder decoder;
     private final Http2ConnectionEncoder encoder;
     private ChannelFutureListener closeListener;
     private BaseDecoder byteDecoder;
-    private static final InternalLogger logger = InternalLoggerFactory.getInstance(Http2ConnectionHandler.class);
 
     public Http2ConnectionHandler(boolean server, Http2FrameListener listener) {
         this(new DefaultHttp2Connection(server), listener);


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 e0447b3e19e..e145b8eb42c 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
@@ -21,28 +21,35 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyShort;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Matchers.isNull;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.reset;
 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 java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
-
-import javax.xml.ws.Holder;
-import java.util.Arrays;
-import java.util.List;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 /**
  * Tests for {@link DefaultHttp2Connection}.
@@ -55,6 +62,9 @@ public class DefaultHttp2ConnectionTest {
     @Mock
     private Http2Connection.Listener clientListener;
 
+    @Mock
+    private Http2Connection.Listener clientListener2;
+
     @Before
     public void setup() {
         MockitoAnnotations.initMocks(this);
@@ -848,6 +858,155 @@ public void circularDependencyWithExclusiveShouldRestructureTree() throws Except
         assertEquals(p.numChildren() * DEFAULT_PRIORITY_WEIGHT, p.totalChildWeights());
     }
 
+    /**
+     * We force {@link #clientListener} methods to all throw a {@link RuntimeException} and verify the following:
+     * <ol>
+     * <li>all listener methods are called for both {@link #clientListener} and {@link #clientListener2}</li>
+     * <li>{@link #clientListener2} is notified after {@link #clientListener}</li>
+     * <li>{@link #clientListener2} methods are all still called despite {@link #clientListener}'s
+     * method throwing a {@link RuntimeException}</li>
+     * </ol>
+     */
+    @Test
+    public void listenerThrowShouldNotPreventOtherListenersFromBeingNotified() throws Http2Exception {
+        final boolean[] calledArray = new boolean[128];
+        // The following setup will ensure that clienListener throws exceptions, and marks a value in an array
+        // such that clientListener2 will verify that is is set or fail the test.
+        int methodIndex = 0;
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onStreamAdded(any(Http2Stream.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onStreamAdded(any(Http2Stream.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onStreamActive(any(Http2Stream.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onStreamActive(any(Http2Stream.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onStreamHalfClosed(any(Http2Stream.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onStreamHalfClosed(any(Http2Stream.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onStreamClosed(any(Http2Stream.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onStreamClosed(any(Http2Stream.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onStreamRemoved(any(Http2Stream.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onStreamRemoved(any(Http2Stream.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onPriorityTreeParentChanged(any(Http2Stream.class), any(Http2Stream.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onPriorityTreeParentChanged(any(Http2Stream.class), any(Http2Stream.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onPriorityTreeParentChanging(any(Http2Stream.class), any(Http2Stream.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onPriorityTreeParentChanging(any(Http2Stream.class), any(Http2Stream.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onWeightChanged(any(Http2Stream.class), anyShort());
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onWeightChanged(any(Http2Stream.class), anyShort());
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onGoAwaySent(anyInt(), anyLong(), any(ByteBuf.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onGoAwaySent(anyInt(), anyLong(), any(ByteBuf.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onGoAwayReceived(anyInt(), anyLong(), any(ByteBuf.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onGoAwayReceived(anyInt(), anyLong(), any(ByteBuf.class));
+
+        doAnswer(new ListenerExceptionThrower(calledArray, methodIndex))
+            .when(clientListener).onStreamAdded(any(Http2Stream.class));
+        doAnswer(new ListenerVerifyCallAnswer(calledArray, methodIndex++))
+            .when(clientListener2).onStreamAdded(any(Http2Stream.class));
+
+        // Now we add clienListener2 and exercise all listener functionality
+        try {
+            client.addListener(clientListener2);
+            Http2Stream stream = client.local().createStream(3);
+            verify(clientListener).onStreamAdded(any(Http2Stream.class));
+            verify(clientListener2).onStreamAdded(any(Http2Stream.class));
+
+            stream.open(false);
+            verify(clientListener).onStreamActive(any(Http2Stream.class));
+            verify(clientListener2).onStreamActive(any(Http2Stream.class));
+
+            stream.setPriority(0, (short) (stream.weight() + 1), true);
+            verify(clientListener).onWeightChanged(any(Http2Stream.class), anyShort());
+            verify(clientListener2).onWeightChanged(any(Http2Stream.class), anyShort());
+            verify(clientListener).onPriorityTreeParentChanged(any(Http2Stream.class),
+                    any(Http2Stream.class));
+            verify(clientListener2).onPriorityTreeParentChanged(any(Http2Stream.class),
+                    any(Http2Stream.class));
+            verify(clientListener).onPriorityTreeParentChanging(any(Http2Stream.class),
+                    any(Http2Stream.class));
+            verify(clientListener2).onPriorityTreeParentChanging(any(Http2Stream.class),
+                    any(Http2Stream.class));
+
+            stream.closeLocalSide();
+            verify(clientListener).onStreamHalfClosed(any(Http2Stream.class));
+            verify(clientListener2).onStreamHalfClosed(any(Http2Stream.class));
+
+            stream.close();
+            verify(clientListener).onStreamClosed(any(Http2Stream.class));
+            verify(clientListener2).onStreamClosed(any(Http2Stream.class));
+            verify(clientListener).onStreamRemoved(any(Http2Stream.class));
+            verify(clientListener2).onStreamRemoved(any(Http2Stream.class));
+
+            client.goAwaySent(client.connectionStream().id(), Http2Error.INTERNAL_ERROR.code(), Unpooled.EMPTY_BUFFER);
+            verify(clientListener).onGoAwaySent(anyInt(), anyLong(), any(ByteBuf.class));
+            verify(clientListener2).onGoAwaySent(anyInt(), anyLong(), any(ByteBuf.class));
+
+            client.goAwayReceived(client.connectionStream().id(),
+                    Http2Error.INTERNAL_ERROR.code(), Unpooled.EMPTY_BUFFER);
+            verify(clientListener).onGoAwayReceived(anyInt(), anyLong(), any(ByteBuf.class));
+            verify(clientListener2).onGoAwayReceived(anyInt(), anyLong(), any(ByteBuf.class));
+        } finally {
+            client.removeListener(clientListener2);
+        }
+    }
+
+    private static final class ListenerExceptionThrower implements Answer<Void> {
+        private static final RuntimeException FAKE_EXCEPTION = new RuntimeException("Fake Exception");
+        private final boolean[] array;
+        private final int index;
+
+        public ListenerExceptionThrower(boolean[] array, int index) {
+            this.array = array;
+            this.index = index;
+        }
+
+        @Override
+        public Void answer(InvocationOnMock invocation) throws Throwable {
+            array[index] = true;
+            throw FAKE_EXCEPTION;
+        }
+    }
+
+    private static final class ListenerVerifyCallAnswer implements Answer<Void> {
+        private final boolean[] array;
+        private final int index;
+
+        public ListenerVerifyCallAnswer(boolean[] array, int index) {
+            this.array = array;
+            this.index = index;
+        }
+
+        @Override
+        public Void answer(InvocationOnMock invocation) throws Throwable {
+            assertTrue(array[index]);
+            return null;
+        }
+    }
+
     private void verifyParentChanging(List<Http2Stream> expectedArg1, List<Http2Stream> expectedArg2) {
         assertSame(expectedArg1.size(), expectedArg2.size());
         ArgumentCaptor<Http2Stream> arg1Captor = ArgumentCaptor.forClass(Http2Stream.class);
@@ -912,18 +1071,18 @@ private void verifyParentChanged(Http2Stream stream, Http2Stream oldParent) {
     }
     private Http2Stream child(Http2Stream parent, final int id) {
         try {
-            final Holder<Http2Stream> streamHolder = new Holder<Http2Stream>();
+            final AtomicReference<Http2Stream> streamReference = new AtomicReference<Http2Stream>();
             parent.forEachChild(new Http2StreamVisitor() {
                 @Override
                 public boolean visit(Http2Stream stream) throws Http2Exception {
                     if (stream.id() == id) {
-                        streamHolder.value = stream;
+                        streamReference.set(stream);
                         return false;
                     }
                     return true;
                 }
             });
-            return streamHolder.value;
+            return streamReference.get();
         } catch (Http2Exception e) {
             PlatformDependent.throwException(e);
             return null;