grpc stream server java.lang.OutOfMemoryError: Direct buffer memory

[wengang]

I run client test code:

        log.info("start query!");
        int count=0;
        while (count++<100) {
            observer.onNext(ChannelData_Query_Request.newBuilder()
                    .addChannelNames("XXXXXX")
                    .setPlant(XXXXXX)
                    .setShotNumber(*******)
                    .setStep(1)
                    .build());
            TimeUnit.MILLISECONDS.sleep(100);
        }
        observer.onCompleted();
        while (!end[0]){
            TimeUnit.SECONDS.sleep(1);
        }

the server fails after 10 successful requests.below is log:

java.lang.OutOfMemoryError: Direct buffer memory
Nov 02 11:49:11 localhost.localdomain java[9835]: at java.base/java.nio.Bits.reserveMemory(Bits.java:175)
Nov 02 11:49:11 localhost.localdomain java[9835]: at java.base/java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:118)
Nov 02 11:49:11 localhost.localdomain java[9835]: at java.base/java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:317)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:758)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:734)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:245)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.netty.buffer.PoolArena.allocate(PoolArena.java:215)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.netty.buffer.PoolArena.allocate(PoolArena.java:147)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:356)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:123)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.netty.shaded.io.grpc.netty.NettyWritableBufferAllocator.allocate(NettyWritableBufferAllocator.java:51)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.MessageFramer.writeRaw(MessageFramer.java:287)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.MessageFramer.access$400(MessageFramer.java:43)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.MessageFramer$OutputStreamAdapter.write(MessageFramer.java:377)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.doFlush(CodedOutputStream.java:3062)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.flushIfNotAvailable(CodedOutputStream.java:3057)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeFixed32NoTag(CodedOutputStream.java:2891)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream.writeFloatNoTag(CodedOutputStream.java:437)
Nov 02 11:49:11 localhost.localdomain java[9835]: at xx.xx.xxx.xxxx.DataType_3D$Data_Y$Data_X.writeTo(DataType_3D.java:388)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeMessageNoTag(CodedOutputStream.java:2855)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeMessage(CodedOutputStream.java:2824)
Nov 02 11:49:11 localhost.localdomain java[9835]: at xx.xx.xxx.xxxx.DataType_3D$Data_Y.writeTo(DataType_3D.java:883)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeMessageNoTag(CodedOutputStream.java:2855)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeMessage(CodedOutputStream.java:2824)
Nov 02 11:49:11 localhost.localdomain java[9835]: at xx.xx.xxx.xxxx.DataType_3D.writeTo(DataType_3D.java:1557)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeMessageNoTag(CodedOutputStream.java:2855)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeMessage(CodedOutputStream.java:2824)
Nov 02 11:49:11 localhost.localdomain java[9835]: at xx.xx.xxx.xxxx.ChannelData.writeTo(ChannelData.java:431)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeMessageNoTag(CodedOutputStream.java:2855)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.CodedOutputStream$OutputStreamEncoder.writeMessage(CodedOutputStream.java:2824)
Nov 02 11:49:11 localhost.localdomain java[9835]: at xx.xx.xxx.xxxx.ChannelData_Query_Respose.writeTo(ChannelData_Query_Respose.java:176)
Nov 02 11:49:11 localhost.localdomain java[9835]: at com.google.protobuf.AbstractMessageLite.writeTo(AbstractMessageLite.java:83)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.protobuf.lite.ProtoInputStream.drainTo(ProtoInputStream.java:52)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.MessageFramer.writeToOutputStream(MessageFramer.java:269)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.MessageFramer.writeKnownLengthUncompressed(MessageFramer.java:230)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.MessageFramer.writeUncompressed(MessageFramer.java:168)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.MessageFramer.writePayload(MessageFramer.java:141)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.AbstractStream.writeMessage(AbstractStream.java:65)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.ServerCallImpl.sendMessageInternal(ServerCallImpl.java:167)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.internal.ServerCallImpl.sendMessage(ServerCallImpl.java:149)
Nov 02 11:49:11 localhost.localdomain java[9835]: at io.grpc.stub.ServerCalls$ServerCallStreamObserverImpl.onNext(ServerCalls.java:365)
Nov 02 11:49:11 localhost.localdomain java[9835]: at xx.xx.xxx.xxxx.processors.PdbChannelProcessor.readWithTimes(PdbChannelProcessor.java:132)
Nov 02 11:49:11 localhost.localdomain java[9835]: at xx.xx.xxx.xxxx.processors.PdbChannelProcessor.read(PdbChannelProcessor.java:46)

[ejona86]

Calling onNext() enqueues the work; the method returning does not imply it has been sent. The API is asynchronous and will generally return before the messages are actually sent. So you're enqueuing a lot of work and then running out memory.

You should cast the StreamObserver passed to the service to ServerCallStreamObserver. You can then use the methods isReady() and setOnReadyHandler() to react to flow control, where you generate messages faster than they are being sent.

[ejona86]

Seems like this is resolved. If not, comment, and it can be reopened.