Repository: netty/netty
Issue: HTTP/2 HEADERS frames need to stay in-place WRT DATA frames.
Labels: defect

Since DATA frames are flow controlled and HEADERS are not, a HEADERS frame sent after DATA (e.g. trailers) can be sent out-of-order due to flow control restrictions.  We need to preserve the original ordering.


Issue comments:
Good catch

Addressed by #3088

Cherry-picked by @nmittler 


Pull request review comments:
Wording is a bit awkward.  Consider clarifying..

Returns the most {@link ChannelFuture} -> Returns the {@link ChannelFuture}

What about continuation frames?  Do we even send continuation frames or are we queuing up header writes until the endofstream happens somewhere?

Is this needed if you are only adding 1 promise object to it?  Why couldn't you just directly use `promise`?

Does this change what we should be returning?  Shouldn't we still return the `ChannelFuture` from the `frameWriter.writeHeaders(...)` call?  I seem to remember returning the promise object during the restructure in a method by mistake and the behavior was undesirable.

I like the consistency and cleanliness of code but what impacts will this have on the majority of time headers are written (the headers before data)?  When will the `operationComplete` be called?  Does this require going back to the event loop?

nit: Consider renaming because of the context we are in.   In the `OutboundFlowState` context this is just the last `newFrame` that was created.

Also for consistency does it make sense to add/use an accessor for this member?

Can you explain this change?

Good catch! I guess we are being good users and not calling header/data methods out of order ;)

Done.

The interface doesn't let you write continuation frames directly.  You write a `Headers` object and that may get broken up into multiple frames.  I don't think this is an issue.

Done. Split out the logic so we only do this when a previous DATA frame was written for the stream.

I believe the code is correct.  We want to return a future that completes when the write completes, which is what will be done in this case.  Once `innerPromise` completes, the aggregator will automatically complete `promise`.

Done ... this has been addressed since I'm not splitting the logic based on whether a previous DATA frame has been sent.

Done.

This should not have been changed ... good catch!

This sounds familiar.  So the Http2Headers must contain all headers for the current headers "phase" (because we don't support an interface to generate continuation frames).


Patch that resolved the issue:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java
index 70f2d6592e7..5e6be2a8a11 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java
@@ -22,6 +22,7 @@
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
+import io.netty.channel.ChannelPromiseAggregator;
 
 import java.util.ArrayDeque;
 
@@ -196,6 +197,11 @@ public void operationComplete(ChannelFuture future) throws Exception {
         return future;
     }
 
+    @Override
+    public ChannelFuture lastWriteForStream(int streamId) {
+        return outboundFlow.lastWriteForStream(streamId);
+    }
+
     @Override
     public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
             boolean endStream, ChannelPromise promise) {
@@ -203,10 +209,12 @@ public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2
     }
 
     @Override
-    public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
-            int streamDependency, short weight, boolean exclusive, int padding, boolean endOfStream,
-            ChannelPromise promise) {
+    public ChannelFuture writeHeaders(final ChannelHandlerContext ctx, final int streamId,
+            final Http2Headers headers, final int streamDependency, final short weight,
+            final boolean exclusive, final int padding, final boolean endOfStream,
+            final ChannelPromise promise) {
         Http2Stream stream = connection.stream(streamId);
+        ChannelFuture lastDataWrite = lastWriteForStream(streamId);
         try {
             if (connection.isGoAway()) {
                 throw protocolError("Sending headers after connection going away.");
@@ -238,6 +246,12 @@ public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2
                                 "Stream %d in unexpected state: %s", stream.id(), stream.state()));
                 }
             }
+
+            if (lastDataWrite != null && !endOfStream) {
+                throw new IllegalStateException(
+                        "Sending non-trailing headers after data has been sent for stream: "
+                                + streamId);
+            }
         } catch (Throwable e) {
             if (e instanceof Http2NoMoreStreamIdsException) {
                 lifecycleManager.onException(ctx, e);
@@ -245,8 +259,49 @@ public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2
             return promise.setFailure(e);
         }
 
+        if (lastDataWrite == null) {
+            // No previous DATA frames to keep in sync with, just send it now.
+            return writeHeaders(ctx, stream, headers, streamDependency, weight, exclusive, padding,
+                    endOfStream, promise);
+        }
+
+        // There were previous DATA frames sent.  We need to send the HEADERS only after the most
+        // recent DATA frame to keep them in sync...
+
+        // Wrap the original promise in an aggregate which will complete the original promise
+        // once the headers are written.
+        final ChannelPromiseAggregator aggregatePromise = new ChannelPromiseAggregator(promise);
+        final ChannelPromise innerPromise = ctx.newPromise();
+        aggregatePromise.add(innerPromise);
+
+        // Only write the HEADERS frame after the previous DATA frame has been written.
+        final Http2Stream theStream = stream;
+        lastDataWrite.addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+                if (!future.isSuccess()) {
+                    // The DATA write failed, also fail this write.
+                    innerPromise.setFailure(future.cause());
+                    return;
+                }
+
+                // Perform the write.
+                writeHeaders(ctx, theStream, headers, streamDependency, weight, exclusive, padding,
+                        endOfStream, innerPromise);
+            }
+        });
+
+        return promise;
+    }
+
+    /**
+     * Writes the given {@link Http2Headers} to the remote endpoint and updates stream state if appropriate.
+     */
+    private ChannelFuture writeHeaders(ChannelHandlerContext ctx, Http2Stream stream,
+            Http2Headers headers, int streamDependency, short weight, boolean exclusive,
+            int padding, boolean endOfStream, ChannelPromise promise) {
         ChannelFuture future =
-                frameWriter.writeHeaders(ctx, streamId, headers, streamDependency, weight,
+                frameWriter.writeHeaders(ctx, stream.id(), headers, streamDependency, weight,
                         exclusive, padding, endOfStream, promise);
         ctx.flush();
 
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2OutboundFlowController.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2OutboundFlowController.java
index e3d55cc7ce4..8afb28a352c 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2OutboundFlowController.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2OutboundFlowController.java
@@ -202,8 +202,14 @@ public ChannelFuture writeData(ChannelHandlerContext ctx, int streamId, ByteBuf
         return promise;
     }
 
+    @Override
+    public ChannelFuture lastWriteForStream(int streamId) {
+        OutboundFlowState state = state(streamId);
+        return state != null ? state.lastNewFrame() : null;
+    }
+
     private static OutboundFlowState state(Http2Stream stream) {
-        return (OutboundFlowState) stream.outboundFlow();
+        return stream != null ? (OutboundFlowState) stream.outboundFlow() : null;
     }
 
     private OutboundFlowState connectionState() {
@@ -383,6 +389,7 @@ final class OutboundFlowState implements FlowState {
         private int pendingBytes;
         private int priorityBytes;
         private int allocatedPriorityBytes;
+        private ChannelFuture lastNewFrame;
 
         private OutboundFlowState(Http2Stream stream) {
             this.stream = stream;
@@ -411,6 +418,13 @@ private int incrementStreamWindow(int delta) throws Http2Exception {
             return window;
         }
 
+        /**
+         * Returns the future for the last new frame created for this stream.
+         */
+        ChannelFuture lastNewFrame() {
+            return lastNewFrame;
+        }
+
         /**
          * Returns the maximum writable window (minimum of the stream and connection windows).
          */
@@ -460,6 +474,8 @@ private int unallocatedPriorityBytes() {
          * Creates a new frame with the given values but does not add it to the pending queue.
          */
         private Frame newFrame(ChannelPromise promise, ByteBuf data, int padding, boolean endStream) {
+            // Store this as the future for the most recent write attempt.
+            lastNewFrame = promise;
             return new Frame(new ChannelPromiseAggregator(promise), data, padding, endStream);
         }
 
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2OutboundFlowController.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2OutboundFlowController.java
index d9b17b8f8c9..a7e4d21c37a 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2OutboundFlowController.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2OutboundFlowController.java
@@ -39,6 +39,12 @@ public interface Http2OutboundFlowController extends Http2DataWriter {
     ChannelFuture writeData(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
             boolean endStream, ChannelPromise promise);
 
+    /**
+     * Returns the {@link ChannelFuture} for the most recent write for the given
+     * stream. If no previous write for the stream has occurred, returns {@code null}.
+     */
+    ChannelFuture lastWriteForStream(int streamId);
+
     /**
      * Sets the initial size of the connection's outbound flow control window. The outbound flow
      * control windows for all streams are updated by the delta in the initial window size. This is


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoderTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoderTest.java
index 1ca69bcb84a..8410ac2939b 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoderTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoderTest.java
@@ -44,6 +44,7 @@
 import io.netty.channel.DefaultChannelPromise;
 
 import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -205,24 +206,55 @@ public void headersWriteAfterGoAwayShouldFail() throws Exception {
 
     @Test
     public void headersWriteForUnknownStreamShouldCreateStream() throws Exception {
-        when(local.createStream(eq(5), eq(false))).thenReturn(stream);
-        encoder.writeHeaders(ctx, 5, EmptyHttp2Headers.INSTANCE, 0, false, promise);
-        verify(local).createStream(eq(5), eq(false));
-        verify(writer).writeHeaders(eq(ctx), eq(5), eq(EmptyHttp2Headers.INSTANCE), eq(0),
+        int streamId = 5;
+        when(stream.id()).thenReturn(streamId);
+        mockFutureAddListener(true);
+        when(local.createStream(eq(streamId), eq(false))).thenReturn(stream);
+        encoder.writeHeaders(ctx, streamId, EmptyHttp2Headers.INSTANCE, 0, false, promise);
+        verify(local).createStream(eq(streamId), eq(false));
+        verify(writer).writeHeaders(eq(ctx), eq(streamId), eq(EmptyHttp2Headers.INSTANCE), eq(0),
                 eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(false), eq(promise));
     }
 
     @Test
     public void headersWriteShouldCreateHalfClosedStream() throws Exception {
-        when(local.createStream(eq(5), eq(true))).thenReturn(stream);
-        encoder.writeHeaders(ctx, 5, EmptyHttp2Headers.INSTANCE, 0, true, promise);
-        verify(local).createStream(eq(5), eq(true));
-        verify(writer).writeHeaders(eq(ctx), eq(5), eq(EmptyHttp2Headers.INSTANCE), eq(0),
+        int streamId = 5;
+        when(stream.id()).thenReturn(5);
+        mockFutureAddListener(true);
+        when(local.createStream(eq(streamId), eq(true))).thenReturn(stream);
+        encoder.writeHeaders(ctx, streamId, EmptyHttp2Headers.INSTANCE, 0, true, promise);
+        verify(local).createStream(eq(streamId), eq(true));
+        verify(writer).writeHeaders(eq(ctx), eq(streamId), eq(EmptyHttp2Headers.INSTANCE), eq(0),
+                eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(true), eq(promise));
+    }
+
+    @Test
+    public void headersWriteAfterDataShouldWait() throws Exception {
+        final AtomicReference<ChannelFutureListener> listener = new AtomicReference<ChannelFutureListener>();
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable {
+                listener.set((ChannelFutureListener) invocation.getArguments()[0]);
+                return null;
+            }
+        }).when(future).addListener(any(ChannelFutureListener.class));
+
+        // Indicate that there was a previous data write operation that the headers must wait for.
+        when(outboundFlow.lastWriteForStream(anyInt())).thenReturn(future);
+        encoder.writeHeaders(ctx, STREAM_ID, EmptyHttp2Headers.INSTANCE, 0, true, promise);
+        verify(writer, never()).writeHeaders(eq(ctx), eq(STREAM_ID), eq(EmptyHttp2Headers.INSTANCE), eq(0),
+                eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(true), eq(promise));
+
+        // Now complete the previous data write operation and verify that the headers were written.
+        when(future.isSuccess()).thenReturn(true);
+        listener.get().operationComplete(future);
+        verify(writer).writeHeaders(eq(ctx), eq(STREAM_ID), eq(EmptyHttp2Headers.INSTANCE), eq(0),
                 eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(true), eq(promise));
     }
 
     @Test
     public void headersWriteShouldOpenStreamForPush() throws Exception {
+        mockFutureAddListener(true);
         when(stream.state()).thenReturn(RESERVED_LOCAL);
         encoder.writeHeaders(ctx, STREAM_ID, EmptyHttp2Headers.INSTANCE, 0, false, promise);
         verify(stream).openForPush();
@@ -233,6 +265,7 @@ public void headersWriteShouldOpenStreamForPush() throws Exception {
 
     @Test
     public void headersWriteShouldClosePushStream() throws Exception {
+        mockFutureAddListener(true);
         when(stream.state()).thenReturn(RESERVED_LOCAL).thenReturn(HALF_CLOSED_LOCAL);
         encoder.writeHeaders(ctx, STREAM_ID, EmptyHttp2Headers.INSTANCE, 0, true, promise);
         verify(stream).openForPush();
@@ -315,6 +348,21 @@ public void settingsWriteShouldNotUpdateSettings() throws Exception {
         verify(writer).writeSettings(eq(ctx), eq(settings), eq(promise));
     }
 
+    private void mockFutureAddListener(boolean success) {
+        when(future.isSuccess()).thenReturn(success);
+        if (!success) {
+            when(future.cause()).thenReturn(new Exception("Fake Exception"));
+        }
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable {
+                ChannelFutureListener listener = (ChannelFutureListener) invocation.getArguments()[0];
+                listener.operationComplete(future);
+                return null;
+            }
+        }).when(future).addListener(any(ChannelFutureListener.class));
+    }
+
     private static ByteBuf dummyData() {
         // The buffer is purposely 8 bytes so it will even work for a ping frame.
         return wrappedBuffer("abcdefgh".getBytes(UTF_8));
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2OutboundFlowControllerTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2OutboundFlowControllerTest.java
index 3fe2e0b2092..20f5314ba52 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2OutboundFlowControllerTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2OutboundFlowControllerTest.java
@@ -29,6 +29,7 @@
 import static org.mockito.Mockito.when;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.http2.DefaultHttp2OutboundFlowController.OutboundFlowState;
@@ -139,6 +140,24 @@ public void frameShouldBeSentImmediately() throws Http2Exception {
         }
     }
 
+    @Test
+    public void lastWriteFutureShouldBeSaved() throws Http2Exception {
+        ChannelPromise promise2 = Mockito.mock(ChannelPromise.class);
+        final ByteBuf data = dummyData(5, 5);
+        try {
+            // Write one frame.
+            ChannelFuture future1 = controller.writeData(ctx, STREAM_A, data, 0, false, promise);
+            assertEquals(future1, controller.lastWriteForStream(STREAM_A));
+
+            // Now write another and verify that the last write is updated.
+            ChannelFuture future2 = controller.writeData(ctx, STREAM_A, data, 0, false, promise2);
+            assertTrue(future1 != future2);
+            assertEquals(future2, controller.lastWriteForStream(STREAM_A));
+        } finally {
+            manualSafeRelease(data);
+        }
+    }
+
     @Test
     public void frameLargerThanMaxFrameSizeShouldBeSplit() throws Http2Exception {
         when(frameWriterSizePolicy.maxFrameSize()).thenReturn(3);
@@ -1174,7 +1193,8 @@ private static int calculateStreamSizeSum(IntObjectMap<Integer> streamSizes, Lis
     }
 
     private void send(int streamId, ByteBuf data, int padding) throws Http2Exception {
-        controller.writeData(ctx, streamId, data, padding, false, promise);
+        ChannelFuture future = controller.writeData(ctx, streamId, data, padding, false, promise);
+        assertEquals(future, controller.lastWriteForStream(streamId));
     }
 
     private void verifyWrite(int streamId, ByteBuf data, int padding) {
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionRoundtripTest.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionRoundtripTest.java
index 6491ced9015..ba9f78b79fc 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionRoundtripTest.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionRoundtripTest.java
@@ -85,8 +85,9 @@ public class Http2ConnectionRoundtripTest {
     private Channel serverChannel;
     private Channel clientChannel;
     private Http2TestUtil.FrameCountDown serverFrameCountDown;
-    private volatile CountDownLatch requestLatch;
-    private volatile CountDownLatch dataLatch;
+    private CountDownLatch requestLatch;
+    private CountDownLatch dataLatch;
+    private CountDownLatch trailersLatch;
 
     @Before
     public void setup() throws Exception {
@@ -106,7 +107,7 @@ public void teardown() throws Exception {
 
     @Test
     public void http2ExceptionInPipelineShouldCloseConnection() throws Exception {
-        bootstrapEnv(1, 1);
+        bootstrapEnv(1, 1, 1);
 
         // Create a latch to track when the close occurs.
         final CountDownLatch closeLatch = new CountDownLatch(1);
@@ -150,7 +151,7 @@ public void listenerExceptionShouldCloseConnection() throws Exception {
                 any(ChannelHandlerContext.class), eq(3), eq(headers), eq(0), eq((short) 16),
                 eq(false), eq(0), eq(false));
 
-        bootstrapEnv(1, 1);
+        bootstrapEnv(1, 1, 1);
 
         // Create a latch to track when the close occurs.
         final CountDownLatch closeLatch = new CountDownLatch(1);
@@ -180,7 +181,7 @@ public void run() {
 
     @Test
     public void nonHttp2ExceptionInPipelineShouldNotCloseConnection() throws Exception {
-        bootstrapEnv(1, 1);
+        bootstrapEnv(1, 1, 1);
 
         // Create a latch to track when the close occurs.
         final CountDownLatch closeLatch = new CountDownLatch(1);
@@ -219,7 +220,7 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
 
     @Test
     public void noMoreStreamIdsShouldSendGoAway() throws Exception {
-        bootstrapEnv(1, 3);
+        bootstrapEnv(1, 3, 1);
 
         // Create a single stream by sending a HEADERS frame to the server.
         final Http2Headers headers = dummyHeaders();
@@ -262,7 +263,7 @@ public Void answer(InvocationOnMock in) throws Throwable {
                 any(ByteBuf.class), eq(0), Mockito.anyBoolean());
         try {
             // Initialize the data latch based on the number of bytes expected.
-            bootstrapEnv(length, 2);
+            bootstrapEnv(length, 2, 1);
 
             // Create the stream and send all of the data at once.
             runInChannel(clientChannel, new Http2Runnable() {
@@ -270,20 +271,25 @@ public Void answer(InvocationOnMock in) throws Throwable {
                 public void run() {
                     http2Client.encoder().writeHeaders(ctx(), 3, headers, 0, (short) 16, false, 0,
                             false, newPromise());
-                    http2Client.encoder().writeData(ctx(), 3, data.retain(), 0, true, newPromise());
+                    http2Client.encoder().writeData(ctx(), 3, data.retain(), 0, false, newPromise());
+
+                    // Write trailers.
+                    http2Client.encoder().writeHeaders(ctx(), 3, headers, 0, (short) 16, false, 0,
+                            true, newPromise());
                 }
             });
 
-            // Wait for all DATA frames to be received at the server.
-            assertTrue(dataLatch.await(5, TimeUnit.SECONDS));
+            // Wait for the trailers to be received.
+            assertTrue(trailersLatch.await(5, TimeUnit.SECONDS));
 
-            // Verify that headers were received and only one DATA frame was received with endStream set.
+            // Verify that headers and trailers were received.
             verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(headers), eq(0),
                     eq((short) 16), eq(false), eq(0), eq(false));
-            verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), any(ByteBuf.class), eq(0),
-                    eq(true));
+            verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(headers), eq(0),
+                    eq((short) 16), eq(false), eq(0), eq(true));
 
             // Verify we received all the bytes.
+            assertEquals(0, dataLatch.getCount());
             out.flush();
             byte[] received = out.toByteArray();
             assertArrayEquals(bytes, received);
@@ -317,9 +323,9 @@ public Void answer(InvocationOnMock in) throws Throwable {
                 return null;
             }
         }).when(serverListener).onDataRead(any(ChannelHandlerContext.class), anyInt(), eq(data),
-                eq(0), eq(true));
+                eq(0), eq(false));
         try {
-            bootstrapEnv(numStreams * text.length(), numStreams * 3);
+            bootstrapEnv(numStreams * text.length(), numStreams * 4, numStreams);
             runInChannel(clientChannel, new Http2Runnable() {
                 @Override
                 public void run() {
@@ -329,18 +335,23 @@ public void run() {
                         http2Client.encoder().writePing(ctx(), false, pingData.slice().retain(),
                                 newPromise());
                         http2Client.encoder().writeData(ctx(), nextStream, data.slice().retain(),
-                                0, true, newPromise());
+                                0, false, newPromise());
+                        // Write trailers.
+                        http2Client.encoder().writeHeaders(ctx(), nextStream, headers, 0,
+                                (short) 16, false, 0, true, newPromise());
                     }
                 }
             });
             // Wait for all frames to be received.
-            assertTrue(requestLatch.await(30, SECONDS));
+            assertTrue(trailersLatch.await(30, SECONDS));
             verify(serverListener, times(numStreams)).onHeadersRead(any(ChannelHandlerContext.class), anyInt(),
                     eq(headers), eq(0), eq((short) 16), eq(false), eq(0), eq(false));
+            verify(serverListener, times(numStreams)).onHeadersRead(any(ChannelHandlerContext.class), anyInt(),
+                    eq(headers), eq(0), eq((short) 16), eq(false), eq(0), eq(true));
             verify(serverListener, times(numStreams)).onPingRead(any(ChannelHandlerContext.class),
                     any(ByteBuf.class));
             verify(serverListener, times(numStreams)).onDataRead(any(ChannelHandlerContext.class),
-                    anyInt(), any(ByteBuf.class), eq(0), eq(true));
+                    anyInt(), any(ByteBuf.class), eq(0), eq(false));
             assertEquals(numStreams, receivedPingBuffers.size());
             assertEquals(numStreams, receivedDataBuffers.size());
             for (String receivedData : receivedDataBuffers) {
@@ -355,9 +366,10 @@ public void run() {
         }
     }
 
-    private void bootstrapEnv(int dataCountDown, int requestCountDown) throws Exception {
+    private void bootstrapEnv(int dataCountDown, int requestCountDown, int trailersCountDown) throws Exception {
         requestLatch = new CountDownLatch(requestCountDown);
         dataLatch = new CountDownLatch(dataCountDown);
+        trailersLatch = new CountDownLatch(trailersCountDown);
         sb = new ServerBootstrap();
         cb = new Bootstrap();
 
@@ -367,7 +379,9 @@ private void bootstrapEnv(int dataCountDown, int requestCountDown) throws Except
             @Override
             protected void initChannel(Channel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
-                serverFrameCountDown = new Http2TestUtil.FrameCountDown(serverListener, requestLatch, dataLatch);
+                serverFrameCountDown =
+                        new Http2TestUtil.FrameCountDown(serverListener, requestLatch, dataLatch,
+                                trailersLatch);
                 p.addLast(new Http2ConnectionHandler(true, serverFrameCountDown));
                 p.addLast(Http2CodecUtil.ignoreSettingsHandler());
             }
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2TestUtil.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2TestUtil.java
index c4f33e00a3b..17fc8adf547 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2TestUtil.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/Http2TestUtil.java
@@ -262,15 +262,22 @@ static class FrameCountDown implements Http2FrameListener {
         private final Http2FrameListener listener;
         private final CountDownLatch messageLatch;
         private final CountDownLatch dataLatch;
+        private final CountDownLatch trailersLatch;
 
         public FrameCountDown(Http2FrameListener listener, CountDownLatch messageLatch) {
             this(listener, messageLatch, null);
         }
 
         public FrameCountDown(Http2FrameListener listener, CountDownLatch messageLatch, CountDownLatch dataLatch) {
+            this(listener, messageLatch, dataLatch, null);
+        }
+
+        public FrameCountDown(Http2FrameListener listener, CountDownLatch messageLatch,
+                CountDownLatch dataLatch, CountDownLatch trailersLatch) {
             this.listener = listener;
             this.messageLatch = messageLatch;
             this.dataLatch = dataLatch;
+            this.trailersLatch = trailersLatch;
         }
 
         @Override
@@ -291,6 +298,9 @@ public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers
                 boolean endStream) throws Http2Exception {
             listener.onHeadersRead(ctx, streamId, headers, padding, endStream);
             messageLatch.countDown();
+            if (trailersLatch != null && endStream) {
+                trailersLatch.countDown();
+            }
         }
 
         @Override
@@ -298,6 +308,9 @@ public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers
                 short weight, boolean exclusive, int padding, boolean endStream) throws Http2Exception {
             listener.onHeadersRead(ctx, streamId, headers, streamDependency, weight, exclusive, padding, endStream);
             messageLatch.countDown();
+            if (trailersLatch != null && endStream) {
+                trailersLatch.countDown();
+            }
         }
 
         @Override