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

@@ -486,6 +486,7 @@ private void streamCreatedInternal(
         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)),

core/src/test/java/io/grpc/internal/ServerImplTest.java

@@ -487,6 +487,7 @@ public void decompressorNotFound() throws Exception {
 
     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();