netty: Abrupt GOAWAY should not cause INTERNAL status

netty/src/main/java/io/grpc/netty/NettyClientHandler.java

@@ -129,6 +129,7 @@ protected void handleNotInUse() {
   private Http2Ping ping;
   private Attributes attributes;
   private InternalChannelz.Security securityInfo;
+  private Status abruptGoAwayStatus;
 
   static NettyClientHandler newHandler(
       ClientTransportLifecycleManager lifecycleManager,
@@ -556,6 +557,21 @@ private void createStream(CreateStreamCommand command, ChannelPromise promise)
       }
       return;
     }
+    if (connection().goAwayReceived()
+        && streamId > connection().local().lastStreamKnownByPeer()) {
+      // This should only be reachable during onGoAwayReceived, as otherwise
+      // getShutdownThrowable() != null
+      command.stream().setNonExistent();
+      Status s = abruptGoAwayStatus;
+      if (s == null) {
+        // Should be impossible, but handle psuedo-gracefully
+        s = Status.INTERNAL.withDescription(
+            "Failed due to abrupt GOAWAY, but can't find GOAWAY details");
+      }
+      command.stream().transportReportStatus(s, RpcProgress.REFUSED, true, new Metadata());
+      promise.setFailure(s.asRuntimeException());
+      return;
+    }
 
     NettyClientStream.TransportState stream = command.stream();
     Http2Headers headers = command.headers();
@@ -772,6 +788,7 @@ public boolean visit(Http2Stream stream) throws Http2Exception {
    */
   private void goingAway(Status status) {
     lifecycleManager.notifyGracefulShutdown(status);
+    abruptGoAwayStatus = status;
     // Try to allocate as many in-flight streams as possible, to reduce race window of
     // https://github.com/grpc/grpc-java/issues/2562 . To be of any help, the server has to
     // gracefully shut down the connection with two GOAWAYs. gRPC servers generally send a PING

netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java

@@ -375,6 +375,23 @@ public void receivedGoAwayShouldNotAffectRacingQueuedStreamId() throws Exception
     assertTrue(future.isDone());
   }
 
+  @Test
+  public void receivedAbruptGoAwayShouldFailRacingQueuedStreamid() throws Exception {
+    // This command has not actually been executed yet
+    ChannelFuture future = writeQueue().enqueue(
+        newCreateStreamCommand(grpcHeaders, streamTransportState), true);
+    // Read a GOAWAY that indicates our stream can't be sent
+    channelRead(goAwayFrame(0, 8 /* Cancel */, Unpooled.copiedBuffer("this is a test", UTF_8)));
+
+    ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(Status.class);
+    verify(streamListener).closed(captor.capture(), same(REFUSED),
+        ArgumentMatchers.<Metadata>notNull());
+    assertEquals(Status.CANCELLED.getCode(), captor.getValue().getCode());
+    assertEquals("HTTP/2 error code: CANCEL\nReceived Goaway\nthis is a test",
+        captor.getValue().getDescription());
+    assertTrue(future.isDone());
+  }
+
   @Test
   public void receivedResetWithRefuseCode() throws Exception {
     ChannelFuture future = enqueue(newCreateStreamCommand(grpcHeaders, streamTransportState));