From ca6fb14791b317022e48e773e7d0629a26459b43 Mon Sep 17 00:00:00 2001
From: Zoe Wang <zoewang@amazon.com>
Date: Thu, 23 Jan 2020 21:26:45 -0800
Subject: [PATCH] Throws `IOException` for the race condition where an http2
 connection gets picked up at the same time it gets closed

---
 .../bugfix-NettyNIOHTTPClient-037932d.json     |  5 +++++
 .../http2/MultiplexedChannelRecord.java        | 10 ++++++----
 .../http2/MultiplexedChannelRecordTest.java    | 18 ++++++++++++++++++
 3 files changed, 29 insertions(+), 4 deletions(-)
 create mode 100644 .changes/next-release/bugfix-NettyNIOHTTPClient-037932d.json

diff --git a/.changes/next-release/bugfix-NettyNIOHTTPClient-037932d.json b/.changes/next-release/bugfix-NettyNIOHTTPClient-037932d.json
new file mode 100644
index 000000000000..fcebf875a6eb
--- /dev/null
+++ b/.changes/next-release/bugfix-NettyNIOHTTPClient-037932d.json
@@ -0,0 +1,5 @@
+{
+    "category": "Netty NIO HTTP Client", 
+    "type": "bugfix", 
+    "description": "Throws `IOException` for the race condition where an HTTP2 connection gets reused at the same time it gets inactive so that failed requests can be retried"
+}
diff --git a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecord.java b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecord.java
index 51868c3420b5..da57f746dcc5 100644
--- a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecord.java
+++ b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecord.java
@@ -28,6 +28,7 @@
 import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.Promise;
 import io.netty.util.concurrent.ScheduledFuture;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -79,7 +80,7 @@ boolean acquireStream(Promise<Channel> promise) {
         return false;
     }
 
-    private void acquireClaimedStream(Promise<Channel> promise) {
+    void acquireClaimedStream(Promise<Channel> promise) {
         doInEventLoop(connection.eventLoop(), () -> {
             if (state != RecordState.OPEN) {
                 String message;
@@ -91,7 +92,7 @@ private void acquireClaimedStream(Promise<Channel> promise) {
                     message = String.format("Connection %s was closed while acquiring new stream.", connection);
                 }
                 log.warn(() -> message);
-                promise.setFailure(new IllegalStateException(message));
+                promise.setFailure(new IOException(message));
                 return;
             }
 
@@ -201,7 +202,7 @@ private void closeAndExecuteOnChildChannels(Consumer<Channel> childChannelConsum
         });
     }
 
-    public void closeAndReleaseChild(Channel childChannel) {
+    void closeAndReleaseChild(Channel childChannel) {
         childChannel.close();
         doInEventLoop(connection.eventLoop(), () -> {
             childChannels.remove(childChannel.id());
@@ -248,9 +249,10 @@ public Channel getConnection() {
         return connection;
     }
 
-    public boolean claimStream() {
+    private boolean claimStream() {
         lastReserveAttemptTimeMillis = System.currentTimeMillis();
         for (int attempt = 0; attempt < 5; ++attempt) {
+
             if (state != RecordState.OPEN) {
                 return false;
             }
diff --git a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecordTest.java b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecordTest.java
index cc5eb8c769f3..115bae593718 100644
--- a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecordTest.java
+++ b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecordTest.java
@@ -16,6 +16,8 @@
 package software.amazon.awssdk.http.nio.netty.internal.http2;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertFalse;
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
@@ -26,8 +28,10 @@
 import io.netty.handler.codec.http2.Http2MultiplexHandler;
 import io.netty.util.concurrent.DefaultPromise;
 import io.netty.util.concurrent.Promise;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -177,6 +181,20 @@ public void availableStream0_reusableShouldBeFalse() {
         assertThat(record.acquireStream(null)).isFalse();
     }
 
+    @Test
+    public void acquireClaimedConnection_channelClosed_shouldThrowIOException() {
+        loopGroup.register(channel).awaitUninterruptibly();
+        Promise<Channel> channelPromise = new DefaultPromise<>(loopGroup.next());
+
+        MultiplexedChannelRecord record = new MultiplexedChannelRecord(channel, 1, Duration.ofSeconds(10));
+
+        record.closeChildChannels();
+
+        record.acquireClaimedStream(channelPromise);
+
+        assertThatThrownBy(() -> channelPromise.get()).hasCauseInstanceOf(IOException.class);
+    }
+
     private EmbeddedChannel newHttp2Channel() {
         EmbeddedChannel channel = new EmbeddedChannel(Http2FrameCodecBuilder.forClient().build(),
                                                       new Http2MultiplexHandler(new NoOpHandler()));