GRPC Java bidirectional streaming

[ankit2995]

My Usecase:

I want to implement bidirectional streaming using java. My usecase has typically 200-2500 messages or requests and each would have an average size of 40kb to be sent to server and recieve the response subsequently. What would be the best approach to send these requests and process responses so that both client and server are not overloaded.Should we batch requests,add some delay,do processing on different threads,etc. Please help with any inputs or approaches based on your experience.Thanks in advance.

[kannanjgithub]

You can use flow control to avoid overwhelming a receiver and also not wait unnecessarily before sending.

[ankit2995]

Thanks @kannanjgithub for you response. Just curious that with this implementation will it not kill the throughput?As in the implementation it will send one request wait for the response to arrive and then only request another 1.

[ankit2995]

If i increase the number of messages to request,i get this error.

io.grpc.StatusRuntimeException: CANCELLED: Failed to read message.
at io.grpc.Status.asRuntimeException(Status.java:533)
at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:481)
at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:489)
at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:453)
at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:486)
at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:574)
at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:72)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:742)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:723)
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:842)
Caused by: io.grpc.StatusRuntimeException: INTERNAL: Invalid protobuf byte sequence
at io.grpc.Status.asRuntimeException(Status.java:525)
at io.grpc.protobuf.lite.ProtoLiteUtils$MessageMarshaller.parse(ProtoLiteUtils.java:240)
at io.grpc.protobuf.lite.ProtoLiteUtils$MessageMarshaller.parse(ProtoLiteUtils.java:134)
at io.grpc.MethodDescriptor.parseResponse(MethodDescriptor.java:284)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInternal(ClientCallImpl.java:667)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInContext(ClientCallImpl.java:654)
... 5 more
Caused by: com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
at com.google.protobuf.InvalidProtocolBufferException.invalidTag(InvalidProtocolBufferException.java:110)
at com.google.protobuf.CodedInputStream$ArrayDecoder.readTag(CodedInputStream.java:607)
at com.fortify.aviator.grpc.AuditorResponse$Builder.mergeFrom(AuditorResponse.java:1006)
at com.fortify.aviator.grpc.AuditorResponse$1.parsePartialFrom(AuditorResponse.java:1994)
at com.fortify.aviator.grpc.AuditorResponse$1.parsePartialFrom(AuditorResponse.java:1986)
at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:63)
at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:25)
at io.grpc.protobuf.lite.ProtoLiteUtils$MessageMarshaller.parseFrom(ProtoLiteUtils.java:245)
at io.grpc.protobuf.lite.ProtoLiteUtils$MessageMarshaller.parse(ProtoLiteUtils.java:237)

[kannanjgithub]

A repro or code snippet will be helpful to diagnose the exception.
About your concern about throughput, yes there is the possibility of reduced throughput with manual flow control if you only request 1 message at a time as given in the example. You can also use a higher value but in order to avoid running out of buffer space you should do it in conjunction with CallStreamObserver.isReady before requesting more as given in the examples.
Note also that manual flow control is for advanced use cases, and gRPC implicitly does automatic flow control via sending acknowledgements to the sender as and when data is read by the receiver.
If you choose not have manual flow control by calling CallStreamObserver.disableAutoRequest(), for advanced use cases one can still use the CallStreamObserver onReadyHandler and stream observer / stream isReady to avoid blocking when the gRPC buffer gets full. The manual flow control examples README mentions these uses.

[kannanjgithub]

Performance should not be a concern when you use onReady handler before writing. If you see a performance difference, they can set onReadyThreshold
CallOptions.withOnReadyThreshold

"Invalid protobuf byte sequence" in the exception you posted means the server is probably calling onNext() from multiple threads and causing corruption.

[ankit2995]

@kannanjgithub Thanks a lot for all your responses. Really appreciate you taking time out and helping me. Also i figured my server was concurrently modifying the stream which caused the above error which you also rightly pointed out. Things are resolved now and i am able to progress. Thanks again.