Direct Buffers vs Heap Memory Occupancy for Queued Streaming Messages

Hi gRPC team,

We are looking for clarification on how gRPC-Java manages memory (Direct Buffers vs. JVM Heap) for incoming streaming messages while they wait in the server executor queue.

Service Profile:
Type: Long-lived Bidirectional / Streaming RPCs
Active Streams: ~15,000 concurrent streams
Transport: ~64 HTTP/2 connections from Envoy to each pod
gRPC Threads: 24
Message Size: Raw wire message is ~500 KB; deserialized Java object is ~5 MB
Memory Budget: 4 GB Heap (-Xmx4096m), ~3 GB Direct/Off-heap memory
Flow Control: Netty BDP tuning enabled (8 MB max window per connection)

Questions:

Memory location while waiting in Executor Queue: Once Netty deframes a message and submits the task to the gRPC server executor, but before a worker thread picks it up and enters onMessage():
1a. Does this queued message reside off-heap in Netty Direct Buffers (raw bytes), or is it already deserialized into the custom Protobuf object on the JVM Heap?
1b. Assuming 1a is true => At what exact point are the underlying Direct Buffers released back to Netty's allocator,
Burst Protection & Direct Memory Sizing (Assuming Off-Heap): Assuming the queued messages reside in Direct Buffers (off-heap, assuming 1a is true) from the above: if there is a sudden surge where ~8,000 streams connect simultaneously and each sends a 500 KB first message, in-flight raw data could reach ~4 GB (8,000 × 500 KB), exceeding our 3 GB Direct Memory limit and risking an off-heap OOM.
2a. Does HTTP/2 connection/stream flow control automatically prevent Direct Memory from growing beyond our 3 GB limit, or could this cause an off-heap OOM?
2b. To protect against this surge, is using a bounded queue on the gRPC executor recommended, or do you recommend any other optimized way for this?

Thanks,
Aditya

@kannanjgithub

Hi @kannanjgithub, as you answered similar questions in this Auto Flow control behaviour for bidirectional streaming
, can you please answer the above queries as well?

Thanks,
Aditya

Great questions! Here is the exact code-verified lifecycle across NettyServerHandler, MessageDeframer, ProtoLiteUtils, and ServerCallImpl:

1a. Where Messages Live While Queued in `callExecutor`

While waiting in callExecutor's queue, each message sits in off-heap Direct Memory (~500 KB), NOT as a 5 MB heap object:

NettyServerHandler.onDataRead slices Netty's pooled DirectByteBuf (data.readRetainedSlice(...)) into a NettyReadableBuffer inside MessageDeframer.
When a full message is assembled, MessageDeframer.deliver() wraps the CompositeReadableBuffer (or lazy decompression stream) in a SingleMessageProducer and enqueues MessagesAvailable onto callExecutor.
Deserialization (call.method.parseRequest(message)) only runs when a callExecutor worker thread executes ServerCallImpl.messagesAvailableInternal(). Thus your concurrent 5 MB Protobuf heap objects are bounded by your 32 active worker threads (32 × 5 MB ≈ 160 MB).

1b. Exact Release Point of `DirectByteBuf`s & Intermediate `byte[]`

Inside parseRequest(message), ProtoLiteUtils.MessageMarshaller.parse(InputStream) copies messages <= 4 MB into a per-worker-thread ThreadLocal<Reference<byte[]>> (~500 KB reusable heap array per worker thread) before parsing.
As CompositeReadableBuffer.execute() reads each slice to 0 remaining bytes, advanceBufferIfNecessary() (CompositeReadableBuffer.java:358-363) immediately calls buffer.close() -> ByteBuf.release() during parseRequest(message), BEFORE listener.onMessage(request) even begins executing (with message.close() at ServerCallImpl.java:347 and GrpcUtil.closeQuietly(producer) cleaning up on any exception).

2a. Why HTTP/2 Flow Control Alone Does NOT Prevent a 4 GB Direct Memory OOM

Look at MessageDeframer.readRequiredBytes() (MessageDeframer.java:362-364):

As soon as the Netty EventLoop deframes Message Update to OkHttp 2.1 final. #1 for a stream and enqueues it into callExecutor, MessageDeframer immediately calls listener.bytesRead(totalBytesRead) -> NettyServerHandler.returnProcessedBytes() on the EventLoop, sending WINDOW_UPDATE back to Envoy before your worker thread dequeues or processes the message!
Because default autoRequest() issues an initial ServerCall.request(1) on all 8,000 streams, all 8,000 streams can simultaneously have 1 deframed 500 KB message sitting in callExecutor's queue (8,000 × 500 KB = 4.0 GB of Direct Memory), plus up to 64 connections × flowControlWindow in MessageDeframer.unprocessed!

2b. Why a Bounded `BlockingQueue` on `callExecutor` Is Unsafe & Recommended Fix

Do NOT use a bounded BlockingQueue on callExecutor: SerializingExecutor does not turn RejectedExecutionException into HTTP/2 backpressure (it logs an error and permanently wedges the stream), and CallerRunsPolicy blocks the Netty EventLoop!
Recommended Fix:
1. Call ServerCallStreamObserver.disableAutoRequest() and gate request(1) behind a global async Semaphore (e.g., 64–128 permits across the server, acquired before request(1) and released at the end of onMessage()). While request(1) is withheld, bytes stay in MessageDeframer.unprocessed without triggering bytesRead() / WINDOW_UPDATE.
2. Set NettyServerBuilder.maxConcurrentCallsPerConnection(...) and .flowControlWindow(...) (which also disables BDP auto-tuning up to 16 MB/conn).

@AgraVator

Thanks for the detailed and insightful explanation @AgraVator ! This clears up a lot of our questions regarding off-heap vs. heap lifecycles.

We have three quick follow-up questions for completeness:

1. JNI & GCLocker in gRPC-Java Internals
We occasionally observe allocation stalls related to the JVM GCLocker during heavy traffic.
From a grpc-java and grpc-netty perspective:
a. Does grpc-java, grpc-netty, or ProtoLiteUtils invoke any JNI critical sections (GetPrimitiveArrayCritical / GetStringCritical) during socket reading, deframing, or copying into ThreadLocal<Reference<byte[]>>?
b. Or is all gRPC off-heap handling implemented via sun.misc.Unsafe.copyMemory and GetDirectBufferAddress (which do not acquire the GCLocker)?

2. Flow Control with Highly Variable / Sparse Stream Periodicity (15s to 2.5 min)
Regarding gating request(1) behind an application Semaphore:

In our service, stream traffic is periodic and variable—some devices transmit every 15 seconds while others send only once every 2.5 minutes.
Because gRPC credits cannot be revoked once granted, calling request(1) on a 2.5-minute stream holds a permit idle for up to 150 seconds. Under a fixed permit pool (e.g., 1,000 permits), idle streams quickly hoard all available permits and starve active streams waiting in the queue.
In scenarios with sparse/variable stream periodicity, is the recommended approach to rely on transport bounds (NettyServerBuilder.flowControlWindow(...) + connection limits) with autoRequest enabled, or is there a recommended gRPC pattern for dynamic credit management?

3. Recommended Approach for Active Stream Tracking & Limiting
To protect against cascading pod failures(due to OOM on some pods) and reject new streams beyond a threshold (e.g., 20,000 active streams):

What is the recommended, most accurate mechanism in gRPC-Java to track active stream lifecycles without race conditions or memory leaks:
1. A ServerInterceptor wrapping ServerCall.Listener (onComplete / onCancel), or
2. A ServerStreamTracer.Factory listening to streamClosed(Status) at the transport layer?

Thanks again for your guidance!

None of grpc-java, grpc-netty or ProtoLiteUtils make JNI calls and off-heap memory handing is indeed handled via Unsafe sun.misc.Unsafe.copyMemory and GetDirectBufferAddress (which do not acquire the GCLocker). The cultprit for JNI method calls is likely gzip if you are using compressed streams. You can verify this by running your repro with profiling enabled.
For handling stream burst loads on Direct memory, reducing the flow control window size and reducing max concurrent streams is the only way. The method suggested of disabling auto flow control and synchronizing and funneling request(int) from different streams can only help with JVM heap memory load, not on Direct memory, because a newly created stream already has the window allowance and will send a window update immediately, untethered to request(int) calls. So if this will cause unfairness for your sparse streams, you don't have to employ this method if your primary concern is avoiding direct memory OOMs.

Edit: The above was written about the initial surge in memory usage for a burst of new streams via NettyServerBuilder.initialFlowControlWindow. After that, for steady-state, request(1) does help reduce both direct memory and heap memory usage.

For active stream tracking & limiting, use ServerStreamTracer.Factory.newServerStreamTracer to track stream creation and ServerStreamTracer.streamClosed to track stream closure. Tracking the count this way at the transport layer is the right way because framework code uses atomics to track stream closures, and this automatically handles
i. Racy cases of the server and the client canceling the call simultaneously (ServerCall.close() and ServerCall.Listener.onCancel() will be invoked and you will need to synchronize for counting)
ii. Streams queued at the transport waiting for the application executor but get cancelled before getting picked up by the executor (in these cases ServerCallListener.onCancel is never called)
iii. Server initiated closes (Listener.onComplete() is never invoked in this cases and you have to wrap ForwardingServerCall.close(...) if you go with the ServerCall way).
You can handle updating the stream counts in the ServerStreamTracer.Factory and access that count and reject new stream creation in the interceptor.

@kannanjgithub

@kannanjgithub thanks a lot, please answer further questions on top of the above.

None of grpc-java, grpc-netty or ProtoLiteUtils make JNI calls and off-heap memory handing is indeed handled via Unsafe sun.misc.Unsafe.copyMemory and GetDirectBufferAddress (which do not acquire the GCLocker). The cultprit for JNI method calls is likely gzip if you are using compressed streams. You can verify this by running your repro with profiling enabled. => on this, yes we have gzip and zstd compressions which is causing GCLocker, just want to check if we have any from grpc side, as it might cause too many retries for the GC, which is not the case as you confirmed.
For handling stream burst loads on Direct memory, reducing the flow control window size and reducing max concurrent streams is the only way. The method suggested of disabling auto flow control and synchronizing and funneling request(int) from different streams can only help with JVM heap memory load, not on Direct memory, because a newly created stream already has the window allowance and will send a window update immediately, untethered to request(int) calls. So if this will cause unfairness for your sparse streams, you don't have to employ this method if your primary concern is avoiding direct memory OOMs. =>

a. We have total of 64 connections per pod from envoy, so the Direct memory limited right by 64 * 16MB(Even with BDP tuning), it's not based on the number of streams right, till request(1) is called by gprc-worker thread when manual flow control is used?
b. If let's say all the streams send at same rate, then gating request (1) with manual flow control should always limit Direct memory right? This is not the case for us though, just checking from understanding point of view.
c. Heap is always limited by grpc-worker-threads * (request deserialized size + processing memory for the request + references to active streams for unprocessed queue/callExecutor) right, as we have 16 grpc-worker-threads so this is always limited right, as even with auto flow control and 15k streams, till grpc-worker-thread picks the next message it's in the raw bytes form stored in Direct memory, so heap is always limited?
d. Reducing the flow control window size and reducing max concurrent streams is the only way for controlling Direct memory => flow control window size has no impact as connections are limited, either we limit total max concurrent streams or total max concurrent sream messages right?

Does the below work for maintaining an accurate stream count, or is an atomic boolean not required if streamClosed is called once per stream always for any scenarios?
`@Component
public class StreamLimitingTracerFactory extends ServerStreamTracer.Factory {

private final AtomicInteger activeStreams = new AtomicInteger(0);

public StreamLimitingTracerFactory() {
Metrics.gauge("grpc.rate.limiter.streams.active", activeStreams, AtomicInteger::get);
}

public int getActiveStreams() {
return activeStreams.get();
}

@OverRide
public ServerStreamTracer newServerStreamTracer(String fullMethodName, Metadata headers) {
activeStreams.incrementAndGet();
return new ServerStreamTracer() {
private final AtomicBoolean closed = new AtomicBoolean(false);
```
     @Override
     public void streamClosed(Status status) {
         if (closed.compareAndSet(false, true)) {
             activeStreams.decrementAndGet();
         }
     }
 };
```
}
}`

Thanks,
Aditya

a. We have total of 64 connections per pod from envoy, so the Direct memory limited right by 64 * 16MB(Even with BDP tuning), it's not based on the number of streams right, till request(1) is called by gprc-worker thread when manual flow control is used?

Flow control is per stream so it would be num-streams * 16MB (with the default window size of 8MB). Manual flow control only lets you control heap memory usage as the deserialization into heap memory is controlled by your application. You should reduce NettyServerBuilder.initialFlowControlWindow as well as NettyServerBuilder.flowControlWindow to accommodate for lesser direct memory availability (-XX:MaxDirectMemorySize for your server).

b.Heap is always limited by grpc-worker-threads * (request deserialized size + processing memory for the request + references to active streams for unprocessed queue/callExecutor) right, as we have 16 grpc-worker-threads so this is always limited right, as even with auto flow control and 15k streams, till grpc-worker-thread picks the next message it's in the raw bytes form stored in Direct memory, so heap is always limited?

Yes

Does the below work for maintaining an accurate stream count, or is an atomic boolean not required if streamClosed is called once per stream always for any scenarios?

Yes, it is strictly guaranteed to be called at most once (and exactly once when the stream terminates)

@kannanjgithub

Thanks @kannanjgithub for the response! We have a couple of follow-up questions regarding your points:

On Flow Control & Direct Memory with 15k Streams:

Flow control is per stream so it would be num-streams * 16MB (with the default window size of 8MB)... You should reduce NettyServerBuilder.initialFlowControlWindow as well as NettyServerBuilder.flowControlWindow to accommodate for lesser direct memory availability => on this response from you, please share your thoughts on the below.

In our setup, we have 15k streams total and process ~400 stream messages per second with 16 gRPC worker threads per pod.

If message processing halts in the application for any reason (e.g., while workers are occupied processing the current 16 stream messages), for the remaining (15k - 16) streams, request(1) would have already been called (under either manual or auto flow control).

In this case, even with a reduced flow control window size, we cannot prevent Direct Memory usage from growing, right? Because the next message for all those (15k - 16) streams will still be received, deframed, and kept in serialized raw bytes form in Direct Memory (in the executor queue), can you please share any solution for limiting Direct Memory for these kinds of scenarios? or gating request(1) with manual flow control and semaphore tokens might fix this, but with a variable stream message rate it would be tricky to handle; also, we need to replenish the token if the stream is closed?

We currently do synchronous processing inside the gRPC worker threads. To prevent Direct Memory bloat and handle downstream (Kafka/DB) stalls with our 15k streams, we are planning to switch to async processing with ordering guarenteed per stream:

Enable manual flow control (disableAutoRequest()).
In onMessage(), store the deserialized request and responseObserver into a separate bounded queue (e.g., size 64) and return immediately.
A separate thread pool consumes from this queue, processes the message (writes to Kafka), and then calls request(1) on that stream's observer.
If the bounded queue is full, we reject and close that stream with RESOURCE_EXHAUSTED.

With this design:
Direct Memory in the gRPC executor queue won't bloat because gRPC threads never block and finish onMessage() immediately (freeing Direct Buffers right away), even if all 15k streams trying to connect at once, we will do rate-limiting so that only a fixed rate of new streams will be allowed, others we will rejct with RESOURCE_EXHAUSTED
Additional Heap Memory will be there for the new bounded queue => bounded queue size * deserialized request size (~5 MB each).
Is this understanding and approach sound and recommended from a grpc-java perspective?

On ServerStreamTracer Guarantees & Atomic Checks:

Yes, it is strictly guaranteed to be called at most once (and exactly once when the stream terminates) => Regarding this:

a. For stream creation (new ServerStreamTracer), you meant it's guaranteed to be called at most once per stream for the stream creation right, but if this is called then it's guaranteed streamClosed to be called exactly once right, no matter of any kind of failure?
b. Why did you called out at-most once for stream creation explicitly, is it even though a new stream is trying to connect, but it may fail before reaching ServerStreamTracer is it, or it's due to any other reason?

Thanks,
Aditya

You are right that the semaphore solution starves slow threads and you need to track stream closures to replenish the token counts. With this approach we can't avoid the limitation of starting slow threads. The rate limiting streams solution you suggested works nicely here.
The at-most once was about stream closed, not about stream creation. I was emphasizing it because the framework's internal synchronization ensures it will never be called more than once even when the client cancels the stream to race with the server's close. And the exactly once part is also true and will happen when the stream does terminate.

@kannanjgithub

Thanks @kannanjgithub for the clarifications! Sorry for the late reply; I was sick for the last few days.

To update on our final architecture, on top of the async onNext and application-level bounded queue we discussed, we are also deferring deserialization to optimize memory, along with rate-limiting new stream creation:

1. Async `onNext` with Deferred Deserialization (Raw Byte Queue)

The Problem with Queuing Deserialized Messages:
Eagerly deserializing upfront expands each ~500 KB wire payload into a ~5 MB Protobuf object graph. Storing deserialized objects in the application bounded queue during downstream latency would balloon heap usage and trigger heavy GC pressure.
Design Approach:
- Custom Byte Marshaller: In bindService(), we register a custom MethodDescriptor with a raw byte marshaller (InputStream -> byte[]). This avoids creating Protobuf objects on the gRPC network thread and immediately releases Netty's off-heap CompositeReadableBuffer, as suggested in this issue
- Raw Byte Queue Handoff: In onNext(byte[] rawBytes), the gRPC thread enqueues the compact raw bytes into an application-level ThreadPoolExecutor (backed by a bounded ArrayBlockingQueue) and returns immediately without blocking.
- Deserialization: Parsing (MetricData.parseFrom(rawBytes)) happens only when an application worker thread dequeues the message to process it.
Benefits: Reduces queued memory footprint by ~90% (holding ~500 KB raw bytes instead of ~5 MB object graphs), keeps deserialized objects short-lived in Young Gen / Eden, and cleanly rejects with RESOURCE_EXHAUSTED if the queue is saturated.

2. Rate-Limiting New Stream Creation

We enforce a token-bucket rate limiter specifically on stream creation (via ServerStreamTracer.Factory). Bursts exceeding the threshold are rejected early with RESOURCE_EXHAUSTED so clients back off and jitter.

Could you please share your thoughts or review on this overall design? Are there any subtleties, edge cases, or caveats from a grpc-java / Netty perspective that we should watch out for with this approach?

Thanks,
Aditya.

About rejecting rate-limited streams

"We enforce a token-bucket rate limiter specifically on stream creation (via ServerStreamTracer.Factory). Bursts exceeding the threshold are rejected early with RESOURCE_EXHAUSTED..."

You Cannot Reject with RESOURCE_EXHAUSTED Directly Inside ServerStreamTracer.Factory. ServerStreamTracer is a read-only observability API:

newServerStreamTracer() does not have access to a ServerCall to close it.
The deprecated serverCallStarted(ServerCall) passes a ReadOnlyServerCall where calling .close() throws UnsupportedOperationException.
If you throw Status.RESOURCE_EXHAUSTED.asRuntimeException() inside newServerStreamTracer(), NettyServerHandler.onHeadersRead() catches it, and converts it into Http2Error.INTERNAL_ERROR, so the client receives Status.INTERNAL(RST_STREAM) instead of Status.RESOURCE_EXHAUSTED.

Use ServerStreamTracer for tracking (or set a Context.Key in filterContext()), but perform the actual call.close(Status.RESOURCE_EXHAUSTED, new Metadata()) inside a ServerInterceptor.

Ensuring optimized GC
By switching from ProtoLiteUtils (which reuses a ThreadLocal<Reference<byte[]>> on the same thread) to passing a newly allocated byte[] per message across threads via ArrayBlockingQueue in your deferred deserialization approach:

With a 4 GB Heap (-Xmx4096m), G1GC defaults to a 1 MB or 2 MB Heap Region Size ($\text{Heap} / 2048$).
In G1GC, any contiguous primitive array $\ge 50%$ of a region size ($\ge 512\text{ KB}$ on a 1 MB region, or $\ge 1\text{ MB}$ on a 2 MB region) is treated as a Humongous Allocation.
Humongous allocations bypass Eden/Young Gen entirely and are allocated directly in Old Gen, which can cause Old Gen fragmentation and premature GC pauses.
Recommendation: Explicitly set -XX:G1HeapRegionSize=4m (or 8m) in JVM flags so that ~500 KB–1 MB byte[] allocations stay well below the 50% humongous threshold and are cleanly allocated in Eden/TLAB.

Avoding Extra Copies in your Marshaller
When implementing MethodDescriptor.Marshaller<byte[]>.parse(InputStream stream), because gRPC’s BufferInputStream implements io.grpc.KnownLength, stream.available() returns the exact byte length of the deframed message upfront. Pre-allocate the exact array size once using int size = stream.available(); byte[] buf = new byte[size]; and read directly into buf (similar to ProtoLiteUtils.java(

grpc-java/protobuf-lite/src/main/java/io/grpc/protobuf/lite/ProtoLiteUtils.java

Lines 191 to 217 in 6f5fcce

    
           if (stream instanceof KnownLength) { 
        
             int size = stream.available(); 
        
             if (size > 0 && size <= DEFAULT_MAX_MESSAGE_SIZE) { 
        
               Reference<byte[]> ref; 
        
               // buf should not be used after this method has returned. 
        
               byte[] buf; 
        
               if ((ref = bufs.get()) == null || (buf = ref.get()) == null || buf.length < size) { 
        
                 buf = new byte[size]; 
        
                 bufs.set(new WeakReference<>(buf)); 
        
               } 
        
               int remaining = size; 
        
               while (remaining > 0) { 
        
                 int position = size - remaining; 
        
                 int count = stream.read(buf, position, remaining); 
        
                 if (count == -1) { 
        
                   break; 
        
                 } 
        
                 remaining -= count; 
        
               } 
        
               if (remaining != 0) { 
        
                 int position = size - remaining; 
        
                 throw new RuntimeException("size inaccurate: " + size + " != " + position); 
        
               } 
        
               cis = CodedInputStream.newInstance(buf, 0, size); 
        
             } else if (size == 0) {

).
Avoid unoptimal draining of the InputStream such as by repeated reading chunk by chunk into a ByteArrayOutputStream and then calling toByteArray on it in the end, for example.
Note that for gzip compressed streams the InputStream passed to your Marshaller won't be implementing io.grpc.KnownLength however, so you need to have a fallback path when the number of bytes is not known as well.

Preserving Per-Stream Ordering & Thread-Safety on responseObserver
With onNext(byte[] rawBytes) handing off to a multi-threaded ThreadPoolExecutor:

Ordering: Per-stream message ordering is only guaranteed if at most 1 message per stream is in the ThreadPoolExecutor at a time—meaning request(1) must only be called at the end of the application worker task (after processing finishes). If you ever pipeline (request(n > 1)), you should wrap the application executor in a per-stream serializing queue/executor.
StreamObserver Thread-Safety: Because onNext(byte[]) (on the gRPC thread) may reject with responseObserver.onError(RESOURCE_EXHAUSTED) when the ArrayBlockingQueue is full, while application worker threads call responseObserver.onNext() / onError() / onCompleted(), all outbound calls on responseObserver must be synchronized per stream (ServerCallStreamObserver is not thread-safe, though request(1) is).

Still Disable Auto-BDP (.flowControlWindow(...)) on NettyServerBuilder
Even with deferred deserialization and disableAutoRequest(), while a stream's current message is sitting in the ArrayBlockingQueue (pendingDeliveries == 0), the client can still pre-send bytes for its next message up to the HTTP/2 stream window into MessageDeframer.unprocessed (Direct Memory).

Make sure .flowControlWindow(...) (e.g., 64 KB – 256 KB) is still configured on NettyServerBuilder so auto-BDP does not expand idle/waiting stream windows to 8 MB in Direct Memory.

@kannanjgithub

Thanks @kannanjgithub.

As we are designing the coordination between our inbound listener (requestObserver), the outbound sender (responseObserver), and our asynchronous application worker queue, we’d appreciate your guidance on a few lifecycle and safety edge cases:

Double close / race between server onError() and transport cancellation: In bidirectional streaming, suppose our worker encounters a business error/connection reset and calls responseObserver.onError(...). Concurrently or immediately after, the client/Envoy disconnects and gRPC dispatches transport cancellation to requestObserver.onError(CANCELLED) (or onCancelHandler).

If responseObserver.onError() is inadvertently called a second time (sequentially), gRPC throws java.lang.IllegalStateException: call already closed from ServerCallImpl.closeInternal.
Similarly, if an in-flight message calls responseObserver.onNext() after the stream aborted, it throws IllegalStateException: Stream was terminated by error.
Question on Safety: When these IllegalStateExceptions are thrown, does gRPC core / Netty still cleanly release underlying Netty transport streams, channel resources, and DirectByteBufs, or could this lead to direct/heap memory leaks?
Question on Best Practice: Is it recommended to guard responseObserver calls with an AtomicBoolean (or at the ServerInterceptor level) to absorb redundant closes, or is there a standard idiom in gRPC-Java for this?
2. Coordination with an asynchronous application queue: When decoupling inbound messages into an internal queue (with disableAutoInboundFlowControl() and manual request(1)):

How should half-close (requestObserver.onCompleted()) be coordinated with worker threads still draining pending tasks in the queue, so that responseObserver.onCompleted() is not called prematurely?
What is the recommended pattern to safely drop pending tasks when isCancelled() becomes true without racing against in-flight onNext() calls?

Thanks,
Aditya

Neither gRPC core nor Netty will leak transport streams, channel resources, or DirectByteBufs when those IllegalStateExceptions are thrown.
To handle races with transport originated cancellations, since you have a bounded queue based processing separate from the server call executor, your worker threads can check for serverCallStreamObserver.isCancelled() before processing a queued item for immediate cancellation visibility.
serverCallStreamObserver.setOnCancelHandler also allows you to get a cancellation callback serialized with other request callbacks (onNext, onError). When this cancellation handler is set, responseObserver.onNext and responseObserver.onErrors called after transport cancellation become no-ops instead of throwing IllegalStateException. It is designed this way because this cancellation callback is queued on the server call executor. For immediate cancellation visibility the aforementioned method can be used.

Using AtomicBoolean is upto you depending on whether multiple threads can touch responseObserver but it does nothing to handle the race with cancellation originating from the transport.

@kannanjgithub

Thanks @kannanjgithub, We want some in-depth clarification on thread-confinement guarantees, task execution ordering, race conditions during stream closure, and fail-safe behavior under unhandled exceptions, can you please answer the below?

Question 1: Synchronous Execution Guarantees Across ServerCall, ServerCall.Listener, Interceptors, and Observers
In a standard gRPC-Java server setup (with worker threads and default SerializingExecutor):

1a. When an inbound message is delivered, does the entire pipeline: Interceptor ServerCall.Listener.onMessage() → requestObserver.onNext() → synchronous responseObserver.onNext()(assuming we call it at the end of requestObserver.onNext() processing synchronously) → Interceptor ServerCall.sendMessage() execute synchronously on the exact same grpc-worker thread within a single call stack frame?
1b. If a client cancels an active stream before sending half-close: Do all interceptor ServerCall.Listener.onCancel() methods, the registered ServerCallStreamObserver.setOnCancelHandler() (if registered), and requestObserver.onError(Status.CANCELLED) execute synchronously and sequentially on the same grpc-worker thread?
1c. When a client half-closes its stream (END_STREAM): Does the chain across all interceptor ServerCall.Listener.onHalfClose() callbacks down to requestObserver.onCompleted() execute synchronously on a single grpc-worker thread?
1d. If a client cancels after half-close has already completed (i.e., requestObserver.onCompleted() was already executed):
Are interceptor ServerCall.Listener.onCancel() and onCancelHandler.run() (if registered) still executed on that grpc-worker thread?
Why is requestObserver.onError() not called here? Is it specifically suppressed because requestObserver.onCompleted() is a terminal state, and delivering an onError afterwards would violate the observer contract?

Question 2: Concurrency, Task Draining, and Manual Flow Control in SerializingExecutor
Regarding the per-stream SerializingExecutor:

2a. In SerializingExecutor.run(), queued tasks are polled in a loop. Does this mean the assigned grpc-worker thread executes all currently queued tasks for that stream sequentially until the stream's queue is completely empty before yielding back to the global thread pool?
2b. If manual flow control is configured with ServerCallStreamObserver.request(2), and 2 messages are already deframed and waiting: When a grpc-worker thread picks up this stream, will both ServerCall.Listener.onMessage() → requestObserver.onNext() calls be executed one after another on the same grpc-worker thread, without interleaving tasks from any other stream on that thread?
2c. How does SerializingExecutor guarantee that no two worker threads ever process tasks for the same stream in parallel, while still allowing different streams to process concurrently across the thread pool?

Question 3: Behavior, Race Conditions, and Lifecycle during responseObserver.onError()
Regarding server-initiated error closures:

3a. When the server initiates closure by calling responseObserver.onError(throwable) (with no prior client cancel or half-close):
3a-i. Does ServerCall.close() execute synchronously on the same grpc-worker thread immediately inside the responseObserver.onError() call to enqueue trailers into Netty's WriteQueue?
3a-ii. Once the wire write is confirmed, is ServerCall.Listener.onComplete() scheduled into the stream's SerializingExecutor and executed asynchronously on a worker thread?
3b. Between the outbound write confirmation and an inbound client cancellation, is the application guaranteed to receive strictly either ServerCall.Listener.onComplete() OR ServerCall.Listener.onCancel(), but never both?
3c. If a client cancellation task is already queued in SerializingExecutor at the moment responseObserver.onError() is called (or if they race): What is the expected behavior if the application attempts to close the call from both paths?
3d. Since gRPC-core and Netty safely tear down and deallocate the transport stream regardless, why does gRPC throw IllegalStateException: call already closed on duplicate close attempts? Is this strictly an intentional contract enforcement to alert users that their application state machine has a race or improper handling?
3e. If transport-level stream teardown in Netty occurs before the final onCancel or onComplete task is scheduled to run: When are the SerializingExecutor, its internal runQueue, and the associated stream tasks eligible for garbage collection? Does full memory reclamation occur only after the final scheduled task finishes executing and the SerializingExecutor drains completely?

Question 4: Exception Handling & Resource Teardown Behavior
If an unhandled exception (RuntimeException or Error) is thrown at various lifecycle points, what is the guaranteed behavior in each case? Specifically:

4a. Initialization / Setup Phase: If an unhandled exception is thrown inside ServerInterceptor.interceptCall() or the initial service implementation method (while returning the requestObserver):
Is the stream immediately terminated with an error to the client, and are direct-memory buffers and stream resources cleanly deallocated?
Does gRPC-Java install a no-op listener (e.g. NOOP_LISTENER) so that any subsequent transport events (such as onCancel or onComplete) are safely absorbed as no-ops without crashing or leaking memory?
4b. Inbound Message Processing: If an unhandled exception is thrown inside ServerCall.Listener.onMessage() or requestObserver.onNext():
Does gRPC forcibly close the stream and send Status.UNKNOWN to the client?
Are any unread/buffered messages immediately discarded and memory reclaimed?
Does the worker thread remain healthy and return to the global pool?
4c. Inbound Completion & Cancellation Callbacks: If an unhandled exception occurs inside:
ServerCall.Listener.onHalfClose() / requestObserver.onCompleted()
ServerCall.Listener.onCancel() / onCancelHandler.run() / requestObserver.onError()
ServerCall.Listener.onComplete()
How does gRPC handle the exception, and is cleanup guaranteed?
4d. Outbound Observer & Call Invocations: If an unhandled exception occurs while calling:
responseObserver.onNext()
responseObserver.onError()
responseObserver.onCompleted()
ServerCall.close()
Does gRPC guarantee that all stream resources, context, and Netty HTTP/2 stream state are torn down properly without hanging or leaking memory?

Thanks,
Aditya

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Repository navigation

Direct Buffers vs Heap Memory Occupancy for Queued Streaming Messages #13066

1a. Where Messages Live While Queued in `callExecutor`

1b. Exact Release Point of `DirectByteBuf`s & Intermediate `byte[]`

2a. Why HTTP/2 Flow Control Alone Does NOT Prevent a 4 GB Direct Memory OOM

2b. Why a Bounded `BlockingQueue` on `callExecutor` Is Unsafe & Recommended Fix

1. Async `onNext` with Deferred Deserialization (Raw Byte Queue)

2. Rate-Limiting New Stream Creation

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	if (stream instanceof KnownLength) {
	int size = stream.available();
	if (size > 0 && size <= DEFAULT_MAX_MESSAGE_SIZE) {
	Reference<byte[]> ref;
	// buf should not be used after this method has returned.
	byte[] buf;
	if ((ref = bufs.get()) == null \|\| (buf = ref.get()) == null \|\| buf.length < size) {
	buf = new byte[size];
	bufs.set(new WeakReference<>(buf));
	}

	int remaining = size;
	while (remaining > 0) {
	int position = size - remaining;
	int count = stream.read(buf, position, remaining);
	if (count == -1) {
	break;
	}
	remaining -= count;
	}

	if (remaining != 0) {
	int position = size - remaining;
	throw new RuntimeException("size inaccurate: " + size + " != " + position);
	}
	cis = CodedInputStream.newInstance(buf, 0, size);
	} else if (size == 0) {

Sunbelt Computer Software

PL/B Language Development and Support

Direct Buffers vs Heap Memory Occupancy for Queued Streaming Messages #13066

Description

Activity

adityapallapolu-dot commented on Sep 24, 2026

AgraVator commented on Sep 28, 2026

1a. Where Messages Live While Queued in callExecutor

1b. Exact Release Point of DirectByteBufs & Intermediate byte[]

2a. Why HTTP/2 Flow Control Alone Does NOT Prevent a 4 GB Direct Memory OOM

2b. Why a Bounded BlockingQueue on callExecutor Is Unsafe & Recommended Fix

adityapallapolu-dot commented on Sep 29, 2026

kannanjgithub commented on Oct 1, 2026

adityapallapolu-dot commented on Oct 1, 2026

kannanjgithub commented on Oct 6, 2026

adityapallapolu-dot commented on Oct 6, 2026

kannanjgithub commented on Oct 8, 2026

adityapallapolu-dot commented on Oct 12, 2026

1. Async onNext with Deferred Deserialization (Raw Byte Queue)

2. Rate-Limiting New Stream Creation

kannanjgithub commented on Oct 15, 2026

adityapallapolu-dot commented on Oct 16, 2026

kannanjgithub commented on Oct 19, 2026

adityapallapolu-dot commented on Oct 19, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

1a. Where Messages Live While Queued in `callExecutor`

1b. Exact Release Point of `DirectByteBuf`s & Intermediate `byte[]`

2b. Why a Bounded `BlockingQueue` on `callExecutor` Is Unsafe & Recommended Fix

1. Async `onNext` with Deferred Deserialization (Raw Byte Queue)