Add best-effort detection of queuing to ThreadlessExecutor after draining complete

Please answer these questions before submitting your issue.

### What version of gRPC are you using?
1.3.0

### What JVM are you using (`java -version`)?
java version "1.8.0_144"
Java(TM) SE Runtime Environment (build 1.8.0_144-b01)
Java HotSpot(TM) 64-Bit Server VM (build 25.144-b01, mixed mode)

### What did you do?
If possible, provide a recipe for reproducing the error.

During use of a specific client, netty begins to detect memory leaks, and my app eventually run out of direct memory. My app eventually begins to receive OutOfDirectMemory errors which occur much later than the leaks are detected and manifest mostly as an immediate UNKNOWN status from the blocking stub. The stack trace of what that looks like is below.

```
at io.grpc.stub.ClientCalls.toStatusRuntimeException(ClientCalls.java:227)
at io.grpc.stub.ClientCalls.getUnchecked(ClientCalls.java:208)
at io.grpc.stub.ClientCalls.blockingUnaryCall(ClientCalls.java:141)
...company specific stack trace removed
Caused by: io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 16777216 byte(s) of direct memory (used: 520093998, max: 536870912)
at io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:624)
at io.netty.util.internal.PlatformDependent.allocateDirectNoCleaner(PlatformDependent.java:578)
at io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:718)
at io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:707)
at io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:239)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:221)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:141)
at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:287)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:170)
at io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:131)
at io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator$MaxMessageHandle.allocate(DefaultMaxMessagesRecvByteBufAllocator.java:73)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:117)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:642)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:565)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:479)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:441)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
... 1 more
```

I'm using multiple gRPC and Netty clients, and even after leaks are detected and even after I see the stack trace above begin to be logged, it appears that these netty and grpc clients will continue to operate normally as long as their connections are maintained. Upon attempting to create new connections, they fail, and the tcp estab errors are capable of being detected at the os layer and show up in metrics dashboards.

Here is the report from the leak detector:

```
ERROR : [grpc-default-worker-ELG-1-84] [] LEAK: ByteBuf.release() was not called before it's garbage-collected. See http://netty.io/wiki/reference-counted-objects.html for more information.
WARNING: 47 leak records were discarded because the leak record count is limited to 4. Use system property io.netty.leakDetection.maxRecords to increase the limit.
Recent access records: 4
#4:
io.netty.buffer.AdvancedLeakAwareByteBuf.readByte(AdvancedLeakAwareByteBuf.java:396)
io.netty.handler.codec.http2.Http2CodecUtil.readUnsignedInt(Http2CodecUtil.java:204)
io.netty.handler.codec.http2.DefaultHttp2FrameReader.processHeaderState(DefaultHttp2FrameReader.java:192)
io.netty.handler.codec.http2.DefaultHttp2FrameReader.readFrame(DefaultHttp2FrameReader.java:148)
io.netty.handler.codec.http2.Http2InboundFrameLogger.readFrame(Http2InboundFrameLogger.java:41)
io.grpc.netty.FixedHttp2ConnectionDecoder.decodeFrame(FixedHttp2ConnectionDecoder.java:119)
io.netty.handler.codec.http2.Http2ConnectionHandler$FrameDecoder.decode(Http2ConnectionHandler.java:341)
io.netty.handler.codec.http2.Http2ConnectionHandler.decode(Http2ConnectionHandler.java:401)
io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:341)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:926)
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:129)
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:642)
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:565)
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:479)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:441)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
java.lang.Thread.run(Thread.java:748)
#3:
io.netty.buffer.AdvancedLeakAwareByteBuf.readUnsignedByte(AdvancedLeakAwareByteBuf.java:402)
io.netty.handler.codec.http2.DefaultHttp2FrameReader.processHeaderState(DefaultHttp2FrameReader.java:191)
io.netty.handler.codec.http2.DefaultHttp2FrameReader.readFrame(DefaultHttp2FrameReader.java:148)
io.netty.handler.codec.http2.Http2InboundFrameLogger.readFrame(Http2InboundFrameLogger.java:41)
io.grpc.netty.FixedHttp2ConnectionDecoder.decodeFrame(FixedHttp2ConnectionDecoder.java:119)
io.netty.handler.codec.http2.Http2ConnectionHandler$FrameDecoder.decode(Http2ConnectionHandler.java:341)
io.netty.handler.codec.http2.Http2ConnectionHandler.decode(Http2ConnectionHandler.java:401)
io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:341)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:926)
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:129)
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:642)
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:565)
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:479)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:441)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
java.lang.Thread.run(Thread.java:748)
#2:
io.netty.buffer.AdvancedLeakAwareByteBuf.readByte(AdvancedLeakAwareByteBuf.java:396)
io.netty.handler.codec.http2.DefaultHttp2FrameReader.processHeaderState(DefaultHttp2FrameReader.java:190)
io.netty.handler.codec.http2.DefaultHttp2FrameReader.readFrame(DefaultHttp2FrameReader.java:148)
io.netty.handler.codec.http2.Http2InboundFrameLogger.readFrame(Http2InboundFrameLogger.java:41)
io.grpc.netty.FixedHttp2ConnectionDecoder.decodeFrame(FixedHttp2ConnectionDecoder.java:119)
io.netty.handler.codec.http2.Http2ConnectionHandler$FrameDecoder.decode(Http2ConnectionHandler.java:341)
io.netty.handler.codec.http2.Http2ConnectionHandler.decode(Http2ConnectionHandler.java:401)
io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:341)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:341)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:349)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:926)
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:129)
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:642)
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:565)
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:479)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:441)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
java.lang.Thread.run(Thread.java:748)
Created at:
io.netty.util.ResourceLeakDetector.track(ResourceLeakDetector.java:229)
io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:296)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:170)
io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:131)
io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator$MaxMessageHandle.allocate(DefaultMaxMessagesRecvByteBufAllocator.java:73)
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:117)
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:642)
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:565)
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:479)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:441)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
java.lang.Thread.run(Thread.java:748)
```
Most interesting to me is io.grpc.netty.FixedHttp2ConnectionDecoder.decodeFrame(FixedHttp2ConnectionDecoder.java:119) in the netty report above, it looks like this method declares it can throw and maybe is and something gets into a weird spot?

The server implementation looks like approximately this:
```
try {
responseObserver.onNext(response);
} catch (IllegalArgumentException e) {
responseObserver.onError(Status.INVALID_ARGUMENT.withCause(e).asException());
} catch (Exception e) {
responseObserver.onError(Status.UNKNOWN.withCause(e).asException());
} finally {
responseObserver.onCompleted();
}
```
I'm wondering if maybe the onCompleted() calls coming in behind an onError() call might race and cause an issue in the header frames getting flushed and then decoded? I plan to move the onCompleted() up into the try block, and give this another attempt, but I'm pretty stumped as to what's happening here and would like some advice.

Thank you in advance.

まず ThreadlessExecutor と draining が完了する箇所を追跡します。issue のスタックトレースは、ClientCalls、FixedHttp2ConnectionDecoder、および Netty の HTTP/2 フレームリーダーに関するコンテキストを提供します。draining 後に best-effort のキューイング検出をどのように観測すべきかを判断し、そのうえで executor が意図した状態を報告し、既存の gRPC クライアントの動作を妨げないことを確認します。

索引モデルが issue の本文から書いたものです。