From c7f69c851bbad8ca5be01393f1d793bf3e9ab6b5 Mon Sep 17 00:00:00 2001
From: Eric Anderson <ejona@google.com>
Date: Thu, 26 Mar 2020 16:31:01 -0700
Subject: [PATCH] core: Add missing setListener if decompressor is unknown

Previously AbstractServerStream would throw an exception which would kill the
RPC with a RST_STREAM. Now the server actually responds with a clean error
message and avoids spamming the logs.

WARNING: Exception in onHeadersRead()
java.lang.IllegalStateException
	at com.google.common.base.Preconditions.checkState(Preconditions.java:495)
	at io.grpc.internal.AbstractStream$TransportState.onStreamAllocated(AbstractStream.java:232)
	at io.grpc.internal.AbstractServerStream$TransportState.onStreamAllocated(AbstractServerStream.java:224)
	at io.grpc.netty.NettyServerHandler.onHeadersRead(NettyServerHandler.java:451)
	at io.grpc.netty.NettyServerHandler.access$900(NettyServerHandler.java:101)
	at io.grpc.netty.NettyServerHandler$FrameListener.onHeadersRead(NettyServerHandler.java:807)
	at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder$FrameReadListener.onHeadersRead(DefaultHttp2ConnectionDecoder.java:373)
---
 core/src/main/java/io/grpc/internal/ServerImpl.java     | 1 +
 core/src/test/java/io/grpc/internal/ServerImplTest.java | 1 +
 2 files changed, 2 insertions(+)

diff --git a/core/src/main/java/io/grpc/internal/ServerImpl.java b/core/src/main/java/io/grpc/internal/ServerImpl.java
index 85d8579323..6e9cb9bf5e 100644
--- a/core/src/main/java/io/grpc/internal/ServerImpl.java
+++ b/core/src/main/java/io/grpc/internal/ServerImpl.java
@@ -486,6 +486,7 @@ public final class ServerImpl extends io.grpc.Server implements InternalInstrume
         String encoding = headers.get(MESSAGE_ENCODING_KEY);
         Decompressor decompressor = decompressorRegistry.lookupDecompressor(encoding);
         if (decompressor == null) {
+          stream.setListener(NOOP_LISTENER);
           stream.close(
               Status.UNIMPLEMENTED.withDescription(
                   String.format("Can't find decompressor for %s", encoding)),
diff --git a/core/src/test/java/io/grpc/internal/ServerImplTest.java b/core/src/test/java/io/grpc/internal/ServerImplTest.java
index d2011f9541..2fe2692bcd 100644
--- a/core/src/test/java/io/grpc/internal/ServerImplTest.java
+++ b/core/src/test/java/io/grpc/internal/ServerImplTest.java
@@ -487,6 +487,7 @@ public class ServerImplTest {
 
     transportListener.streamCreated(stream, "Waiter/nonexist", requestHeaders);
 
+    verify(stream).setListener(isA(ServerStreamListener.class));
     verify(stream).streamId();
     verify(stream).close(statusCaptor.capture(), any(Metadata.class));
     Status status = statusCaptor.getValue();