Error logged from client-closed barrage stream

[niloc132]

This appears to be a race condition where the client has already closed the stream, but the server is still trying to write a message to it.

        at com.google.common.base.Preconditions.checkState(Preconditions.java:502)
        at io.grpc.internal.ServerCallImpl.sendMessageInternal(ServerCallImpl.java:162)
        at io.grpc.internal.ServerCallImpl.sendMessage(ServerCallImpl.java:154)
        at io.grpc.ForwardingServerCall.sendMessage(ForwardingServerCall.java:32)
        at io.grpc.stub.ServerCalls$ServerCallStreamObserverImpl.onNext(ServerCalls.java:380)
        at io.deephaven.server.barrage.BarrageStreamGenerator.processBatches(BarrageStreamGenerator.java:770)
        at io.deephaven.server.barrage.BarrageStreamGenerator$SubView.forEachStream(BarrageStreamGenerator.java:380)
        at io.deephaven.server.arrow.ArrowModule$1.onNext(ArrowModule.java:46)
        at io.deephaven.server.arrow.ArrowModule$1.onNext(ArrowModule.java:41)
        at io.deephaven.server.barrage.BarrageMessageProducer.propagateToSubscribers(BarrageMessageProducer.java:1546)
        at io.deephaven.server.barrage.BarrageMessageProducer.updateSubscriptionsSnapshotAndPropagate(BarrageMessageProducer.java:1456)
        at io.deephaven.server.barrage.BarrageMessageProducer$UpdatePropagationJob.run(BarrageMessageProducer.java:1025)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
        at java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at io.deephaven.server.runner.DeephavenApiServerModule$ThreadFactory.lambda$newThread$0(DeephavenApiServerModule.java:164)
        at java.lang.Thread.run(Thread.java:833)

Originally posted by @rcaudy in #3162 (comment)

It might not be appropriate to wrap the entire onNext in a safelyExecuteLocked instead of just a synchronized block, as some errors are expected to escape., though this would let us remove the try/catch in io.deephaven.server.barrage.BarrageMessageProducer#propagateToSubscribers. Instead, we probably would prefer to only handle the case where IO with the grpc stream has failed, close the stream, and not log it, perhaps by wrapping the lambda passed to forEachStream in io.deephaven.server.arrow.ArrowModule#provideListenerAdapter to catch any error that comes from the delegate observer.

Perhaps gRPC anticipates some other pattern to ensure that application code can never write to a closed stream observer?