[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

Commencez par la boucle handleArrowIterator et les vérifications listener.isCancelled() et listener.isReady(), puis reproduisez l’interruption du client face au serveur FlightRPC. Suivez l’impact de l’exception de stream fermé sur ces états du listener ; le travail est terminé lorsque le serveur arrête la boucle et gère le stream client fermé sans rester bloqué.

Rédigé par le modèle d'indexation à partir du texte de l'issue.