Merge pull request #1557 from dagnir/h2-flush-before-read

Flush parent before reading on H2 connection

Author: dagnir

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/HandlerRemovingChannelPool.java

@@ -27,6 +27,7 @@
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Promise;
 import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.http.nio.netty.internal.http2.FlushOnReadHandler;
 
 /**
  * Removes any per request {@link ChannelHandler} from the pipeline prior to releasing
@@ -79,6 +80,7 @@ private void removePerRequestHandlers(Channel channel) {
             removeIfExists(channel.pipeline(),
                            HttpStreamsClientHandler.class,
                            LastHttpContentHandler.class,
+                           FlushOnReadHandler.class,
                            ResponseHandler.class,
                            ReadTimeoutHandler.class,
                            WriteTimeoutHandler.class);

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/NettyRequestExecutor.java

@@ -62,6 +62,7 @@
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.http.Protocol;
+import software.amazon.awssdk.http.nio.netty.internal.http2.FlushOnReadHandler;
 import software.amazon.awssdk.http.nio.netty.internal.http2.Http2ToHttpInboundAdapter;
 import software.amazon.awssdk.http.nio.netty.internal.http2.HttpToHttp2OutboundAdapter;
 import software.amazon.awssdk.http.nio.netty.internal.utils.ChannelUtils;
@@ -172,6 +173,9 @@ private boolean tryConfigurePipeline() {
         }
 
         pipeline.addLast(LastHttpContentHandler.create());
+        if (Protocol.HTTP2.equals(protocol)) {
+            pipeline.addLast(FlushOnReadHandler.getInstance());
+        }
         pipeline.addLast(new HttpStreamsClientHandler());
         pipeline.addLast(ResponseHandler.getInstance());
 

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/FlushOnReadHandler.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.nio.netty.internal.http2;
+
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+
+/**
+ * This is an HTTP/2 related workaround for an issue where a WINDOW_UPDATE is
+ * queued but not written to the socket, causing a read() on the channel to
+ * hang if the remote endpoint thinks our inbound window is 0.
+ */
+@SdkInternalApi
+@ChannelHandler.Sharable
+public final class FlushOnReadHandler extends ChannelOutboundHandlerAdapter {
+    private static final FlushOnReadHandler INSTANCE = new FlushOnReadHandler();
+
+    private FlushOnReadHandler() {
+    }
+
+    @Override
+    public void read(ChannelHandlerContext ctx) {
+        //Note: order is important, we need to fire the read() event first
+        // since it's what triggers the WINDOW_UPDATE frame write
+        ctx.read();
+        ctx.channel().parent().flush();
+    }
+
+    public static FlushOnReadHandler getInstance() {
+        return INSTANCE;
+    }
+}

http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/http2/FlushOnReadTest.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.nio.netty.internal.http2;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.when;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class FlushOnReadTest {
+
+    @Mock
+    private ChannelHandlerContext mockCtx;
+
+    @Mock
+    private Channel mockChannel;
+
+    @Mock
+    private Channel mockParentChannel;
+
+    @Test
+    public void read_forwardsReadBeforeParentFlush() {
+        when(mockCtx.channel()).thenReturn(mockChannel);
+        when(mockChannel.parent()).thenReturn(mockParentChannel);
+
+        FlushOnReadHandler handler = FlushOnReadHandler.getInstance();
+
+        handler.read(mockCtx);
+
+        InOrder inOrder = Mockito.inOrder(mockCtx, mockParentChannel);
+
+        inOrder.verify(mockCtx).read();
+        inOrder.verify(mockParentChannel).flush();
+    }
+
+    @Test
+    public void getInstance_returnsSingleton() {
+        assertThat(FlushOnReadHandler.getInstance() == FlushOnReadHandler.getInstance()).isTrue();
+    }
+}

services/kinesis/src/it/java/software/amazon/awssdk/services/kinesis/SubscribeToShardIntegrationTest.java

@@ -17,6 +17,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import io.reactivex.Flowable;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
@@ -37,6 +38,8 @@
 import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.core.async.SdkPublisher;
 import software.amazon.awssdk.http.SdkHttpResponse;
+import software.amazon.awssdk.http.nio.netty.Http2Configuration;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
 import software.amazon.awssdk.services.kinesis.model.ConsumerStatus;
 import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
 import software.amazon.awssdk.services.kinesis.model.Record;
@@ -82,6 +85,38 @@ public void tearDown() {
                                        .enforceConsumerDeletion(true)).join();
     }
 
+    @Test
+    public void subscribeToShard_smallWindow_doesNotTimeOutReads() {
+        // We want sufficiently large records (relative to the initial window
+        // size we're choosing) so the client has to send multiple
+        // WINDOW_UPDATEs to receive them
+        for (int i = 0; i < 16; ++i) {
+            putRecord(64 * 1024);
+        }
+
+        KinesisAsyncClient smallWindowAsyncClient = KinesisAsyncClient.builder()
+                .credentialsProvider(CREDENTIALS_PROVIDER_CHAIN)
+                .httpClientBuilder(NettyNioAsyncHttpClient.builder()
+                    .http2Configuration(Http2Configuration.builder()
+                            .initialWindowSize(16384)
+                            .build()))
+                .build();
+
+        try {
+            smallWindowAsyncClient.subscribeToShard(r -> r.consumerARN(consumerArn)
+                            .shardId(shardId)
+                            .startingPosition(s -> s.type(ShardIteratorType.TRIM_HORIZON)),
+                    SubscribeToShardResponseHandler.builder()
+                            .onEventStream(es -> Flowable.fromPublisher(es).forEach(e -> {}))
+                            .onResponse(this::verifyHttpMetadata)
+                            .build())
+                    .join();
+
+        } finally {
+            smallWindowAsyncClient.close();
+        }
+    }
+
     @Test
     public void subscribeToShard_ReceivesAllData() {
         List<SdkBytes> producedData = new ArrayList<>();
@@ -178,14 +213,25 @@ private void waitForStreamToBeActive() {
               .orFailAfter(Duration.ofMinutes(5));
     }
 
+
     /**
      * Puts a random record to the stream.
      *
      * @return Record data that was put.
      */
     private Optional<SdkBytes> putRecord() {
+        return putRecord(50);
+    }
+
+    /**
+     * Puts a random record to the stream.
+     *
+     * @param len The number of bytes to generate for the record.
+     * @return Record data that was put.
+     */
+    private Optional<SdkBytes> putRecord(int len) {
         try {
-            SdkBytes data = SdkBytes.fromByteArray(RandomUtils.nextBytes(50));
+            SdkBytes data = SdkBytes.fromByteArray(RandomUtils.nextBytes(len));
             asyncClient.putRecord(PutRecordRequest.builder()
                                                   .streamName(streamName)
                                                   .data(data)