[Java] Force an outage of the client,server stuck in an endless loop

### Describe the usage question you have. Please include as many useful details as possible.

```
try (BufferAllocator allocator = new RootAllocator()) {
// Server
try (final CookbookProducer producer = new CookbookProducer(allocator, location);
final FlightServer flightServer = FlightServer.builder(allocator, location, producer).build()) {
try {
flightServer.start();
System.out.println("S1: Server (Location): Listening on port " + flightServer.getPort());
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
```

And I handle mysql resultset like this:
```
ResultSet resultSet = statement.executeQuery();
ArrowVectorIterator iterator = JdbcToArrow.sqlToArrowVectorIterator(
resultSet, config)
```

```
public void handleArrowIterator(ArrowVectorIterator iterator, BufferAllocator allocator) {
int index = 0;

while (iterator.hasNext() && !listener.isCancelled()) {
if (listener.isReady()) {
try (VectorSchemaRoot root = iterator.next()) {
index++;
VectorUnloader unloader = new VectorUnloader(root);

ArrowRecordBatch arb = unloader.getRecordBatch();

loader.load(arb);
listener.putNext();
}

}

}
listener.completed();
}
```

But when I force an outage of the client,listener.isCancelled() and listener.isReady() always false,The server is stuck in an endless loop.How to reslove it?
Server exception message:
```
io.grpc.netty.NettyServerHandler -Stream Error
io.netty.handler.codec.http2.Http2Exception$StreamException: Stream closed before write could take place
at io.netty.handler.codec.http2.Http2Exception.streamError(Http2Exception.java:172)
at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$FlowState.cancel(DefaultHttp2RemoteFlowController.java:481)
at io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController$1.onStreamClosed(DefaultHttp2RemoteFlowController.java:105)
at io.netty.handler.codec.http2.DefaultHttp2Connection.notifyClosed(DefaultHttp2Connection.java:357)
at io.netty.handler.codec.http2.DefaultHttp2Connection$ActiveStreams.removeFromActiveStreams(DefaultHttp2Connection.java:1007)
at io.netty.handler.codec.http2.DefaultHttp2Connection$ActiveStreams.deactivate(DefaultHttp2Connection.java:963)
at io.netty.handler.codec.http2.DefaultHttp2Connection$DefaultStream.close(DefaultHttp2Connection.java:515)
at io.netty.handler.codec.http2.DefaultHttp2Connection.close(DefaultHttp2Connection.java:153)
at io.netty.handler.codec.http2.Http2ConnectionHandler$BaseDecoder.channelInactive(Http2ConnectionHandler.java:209)
at io.netty.handler.codec.http2.Http2ConnectionHandler.channelInactive(Http2ConnectionHandler.java:417)
at io.grpc.netty.NettyServerHandler.channelInactive(NettyServerHandler.java:628)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1405)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248)
at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:901)
at io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:813)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:164)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)
at io.netty.channel.nio.NioEventLoop.run(NioE
```

### Component(s)

FlightRPC

Bắt đầu với vòng lặp handleArrowIterator và các kiểm tra listener.isCancelled() và listener.isReady(), sau đó tái hiện sự cố gián đoạn của client với server FlightRPC. Theo dõi cách exception về stream đã đóng ảnh hưởng đến các trạng thái listener đó; hoàn tất khi server dừng vòng lặp và xử lý stream client đã đóng mà không tiếp tục bị mắc kẹt.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.