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

1a. Yes
1b. Yes
1c. Yes
1d. Yes and Yes. Because requestObserver.onError() cannot be called when cancellation arrives after onHalfClose(), requestObserver.onError() is not a guaranteed RPC-cancellation callback—which is why ServerCallStreamObserver.setOnCancelHandler(...) exists to notify the server even after half-close.

2a. Yes. With your proopsed manual flow control and requesting 1 message at a time after processing a message, there will atmost be only 1 MessagesAvailable task will be in runQueue at a time anyway.
2b. Yes
2c. Serialization within a stream is achieved by having a dedicated SerializingExecutor per stream wrapping around the shared server call executor, which also allows concurrent messages from different streams to be processed in parallel.

3a-i. Yes
3a-ii. Yes
3b. Yes
3c. If the application only closes responseObserver from its worker thread error path and does not call responseObserver.onError from inside requestObserver.onError(CANCELLED) / onCancelHandler, then there is no IllegalStateException thrown. IllegalStateException is only thrown when your application code calls responseObserver.onError from both paths.
3d. gRPC throws and not silently ignore duplicate closes because, it is necessary to arrive at a single unambiguous close(Status) call for the ServerCall.Listener. Also because StreamObserver is not thread-safe.
3e. Yes, memory reclamation happens only after the final closing task is executed for the stream that closes the Deframer.

4a. Immediate stream termination & resource cleanup: Yes
No-op listener installation: Yes
4b. Unread/buffered messages immediately closed: Yes
Stream forcibly closed with Status.UNKNOWN: Yes
Worker thread healthy and returned to the pool: Yes
4c. The Netty HTTP/2 stream and MessageDeframer have already been torn down and deallocated on the Netty EventLoop thread by the time the completion/cancellation callbacks are run.
4d. When called from the grpc framework callstack, Yes. But if these runtime exceptions happen during the async processing on your kafka worker pool, then your application code must handle the exceptions and close the server call, since ServerCallImpl code isn't automatically reached.

@kannanjgithub

Thanks @kannanjgithub, can you please share your thoughts on the below?

The below queries are with requestObserver.onNext being executed directly on the grpc-worker thread fully and then responseObserver.onNext called at the end with the processed response. Currently we are using lognet global interceptor which is handling exceptions and double serverCall.close. We have to remove this because lognet global interceptor is storing last incoming request in memory throughout the lifecycle of the stream, which is adding to heap memory footprint for us.

When requestObserver.onError is triggered, then we shouldn't call responseObserver.onError inside requestObserver.onError right, as the stream in both directions is already closed when requestObserver.onError is called?
2a. When we get any processing failure inside requestObserver.onNext, we call responseObserver.onError and then if requestObserver.onCompleted is executed we have to call responseObserver.onCompleted(to close the stream, because requestObserver.onCompleted can be called independently too), with this serverCall.close is triggered twice right one with processign failure responseObserver.onError and one with resposneObserver.onCompleted, causing call is closed exception how to solve these?, if not handled does it cause any resources leak?, is this different for unary vs bi-directional?
2b. Due to processing exception in requestObserver.onNext, and called responseObserver.onError, now requestObserver.onNext returns gracefully which calls request(1), and next message might be processed again right, if this message processing doesn't get any error, and proceeds to call responseObserver.onNext, which is causing illegal state exceptions, is there any standard way to prevent this illegal state exceptions? if not handled does it cause any resources leak? is this different for unary vs bi-directional?
2c. Also in addition to the above, if we get any unhandled exception while 2a/2b is happening, it would be difficult to predict the illegal state exceptions, is there any standard way to tackle these? is this different for unary vs bi-directional?
2d. From application side if we guarentee all responseObserver.onNext/onError/onCompleted/serverCall.close/serverCall.sendMessage to be executed synchronously with the requestObserver.onNext/onError/onCompleted/listener.onCancel/listener.onHalfClose methods, then can we safely ignore these illegal state exceptions, or is it better to handle them? is this different for unary vs bi-directional?
Transport level closure of the stream in both directions happens at once, either with the client cancel or after completion of the end trailers write, due to responseObserver.onError/onCompleted right?
Assuming the above is true, Transport level stream closure happened due to client cancel or after completion of the end trailers write due to responseObserver.onError/onCompleted.
a. Stream memory is cleaned up after all pending serializingExecutor tasks are completed right?, is this different for unary vs bi-directional?
b. While executing serializingExecutor tasks even if any exception is caused, still the serializingExecutor queue is drained off completely and grpc-worker-thread executing this still alive right after the completion of all the tasks of serialized executor?, is this different for unary vs bi-directional?
TracerFactory usage is marked as experimental, is it safe to use for maintaining active streams?, what do you mean by experimental here, is it just that implementation might change in future, not the core API or is it something else?
On the below reply from you earlier
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.
6a. 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) => For server initiated closes, listener.onComplete will be called right, but not in the same server close execution right, once the end trailers is written before client cancel happens?, is this different for unary vs bi-directional?
6b. For any stream either listener.onCancel or listener.onComplete should be called right, so we can decrement the counter here right if we use interceptor way to track the active stream count. Just asking from understanding point of view and we don't have any racy case also right, as either listener.onCancel or listener.onComplete is called for any stream if init is successful?, is this different for unary vs bi-directional?
6c. 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) => This case will be missing in the active streams tracking if handling from interceptor, but the count is neither increased nor decremented right, so active stream tracking is eventually consistent?, is this different for unary vs bi-directional?
6d. Till the interceptor methods get called we have higher active streams count if tracking via interceptor, but with TracerFactory the decrement of dead streams is faster right, is this the major benefit to use TracerFactory instead of interceptor tracking?, is this different for unary vs bi-directional?

Thanks,
Aditya

@kannanjgithub

Hi @kannanjgithub,

We want to verify our understanding of gRPC-Java’s internal lifecycle, exception semantics, and best practices in bidirectional streaming when message processing runs synchronously on the gRPC worker thread (via SerializingExecutor, without asynchronous application queues case).

Could you please confirm or correct our understanding for each numbered point below?

0. Foundational Stream Resource Cleanup & Worker Thread Resilience Guarantee
0.1. Can you confirm that no matter what exception occurs throughout the lifecycle of a stream (in business logic, serialization, or transport writes, during excution of responseObserver.onNext/onError/onCompleted), as long as calls to responseObserver.onNext(), responseObserver.onError(), and responseObserver.onCompleted() are invoked in a thread-safe manner, complete stream resource cleanup (Netty HTTP/2 stream state, framing buffers, deframers) is always guaranteed with zero resource leak, and the underlying grpc-server-* worker thread never dies right?

1. Interceptor Exception Safety & Delegation
1.1. In interceptors, is it the recommended best practice to isolate all custom logic (metrics, logging, tracing, MDC) with try / catch (Exception)—allowing fatal java.lang.Errors like OutOfMemoryError to propagate—and always invoke delegate calls (super.close, super.sendMessage, super.onMessage, super.onCancel, super.onHalfClose, super.onComplete) in finally, ensuring observability failures never escape to override the application's intended status with Status.UNKNOWN?

2. Stream Lifecycle: Admission Control vs. Mid-Stream Termination
2.1. Admission control: Is calling ServerCall.close(STATUS, trailers) from an interceptor only appropriate before next.startCall() (e.g. auth check or rate-limiting during interceptCall), and does returning a no-op listener new ServerCall.Listener() {} guarantee that downstream interceptors and service code are never executed (with no requestObserver or responseObserver created)?

2.2. Mid-stream ServerCall.close() anti-pattern: Once a stream is active, is calling ServerCall.close() directly from an interceptor an anti-pattern because it silently marks closeCalled = true in core without notifying responseObserver (and without setting isCancelled(), isCancelled will be set for client closures), inevitably causing subsequent application responseObserver.onNext() calls to crash with IllegalStateException: call is closed and responseObserver.onError() / responseObserver.onCompleted() calls to crash with IllegalStateException: call already closed?

2.3. Cooperative cancellation: If an interceptor needs to terminate an active stream mid-stream (e.g. idle timeout or max connection lifetime), is cancelling Context.CancellableContext (e.g. via withDeadlineAfter or cancellableContext.cancel()) the idiomatic approach to trigger client RST_STREAM, listener.onCancel(), serverCallObserver.setOnCancelHandler(), and requestObserver.onError(Status.CANCELLED) cleanly without IllegalStateException crashes?

3. Outbound Exception Semantics: responseObserver.onError() and responseObserver.onCompleted()
3.1. Asynchronous transport decoupling: Do responseObserver.onError() and responseObserver.onCompleted() decouple from the network by enqueuing trailers asynchronously onto Netty's WriteQueue, guaranteeing they never throw synchronous I/O, socket, or network exceptions?

3.2. Single vs. double close: Barring passing a literal null to responseObserver.onError(null), is IllegalStateException: call already closed the only runtime exception responseObserver.onError() and responseObserver.onCompleted() can ever throw back to the caller, occurring strictly on duplicate close calls?

4. Outbound Exception Semantics: responseObserver.onNext()
4.1. Pre-send state checks: Before attempting to send, does responseObserver.onNext() strictly throw only StatusRuntimeException: CANCELLED (unless onCancelHandler is registered and prior client close), IllegalStateException: Stream was terminated by error (if responseObserver.onError() was called earlier), or IllegalStateException: Stream is already completed (if responseObserver.onCompleted() was called earlier)?

4.2. Swallowed write RuntimeException: In ServerCallImpl.sendMessageInternal(), if a RuntimeException occurs during Protobuf serialization or transport writing (writeMessage/flush), does gRPC catch it, invoke ServerCall.close() internally (setting closeCalled = true), and swallow it without rethrowing back to the caller (rethrowing only fatal java.lang.Error)?

4.3. Inevitable secondary exceptions: Because of this silent closure case from 4.2, will subsequent calls on that stream inevitably throw IllegalStateException: call is closed on the next responseObserver.onNext() call, or IllegalStateException: call already closed on subsequent responseObserver.onError() / responseObserver.onCompleted() calls?

5. Application Solution: The OutboundResponse Wrapper Pattern
In bidirectional streaming, stream closure can be initiated from multiple independent application callbacks:

An error during message processing in requestObserver.onNext() (calling responseObserver.onError()).
Client half-close in requestObserver.onCompleted() (calling responseObserver.onCompleted()).
Client cancellation in requestObserver.onError() (calling responseObserver.onError()).
Mid-stream application timeouts or connection resets.
Furthermore, gRPC provides no public isClosed() method to proactively detect silent transport write closures (as in 4.2). Therefore, we wrap responseObserver in an application-level wrapper:

private final class OutboundResponse {

private final StreamObserver<T> responseObserver;
private boolean closed; // or volatile boolean closed

private OutboundResponse(StreamObserver<T> responseObserver) {
    this.responseObserver = responseObserver;
}

public void onNext(T message) {
    if (closed) {
        return;
    }
    try {
        responseObserver.onNext(message);
    } catch (IllegalStateException e) {
        // Catches:
        // 1. "call is closed" (if an earlier transport write failure silently closed the call)
        // 2. "Stream was terminated by error" (if onError was called earlier)
        // 3. "Stream is already completed" (if onCompleted was called earlier)
        closed = true;
    }
}

public void onError(Throwable error) {
    if (closed) {
        return;
    }
    closed = true;
    try {
        responseObserver.onError(error);
    } catch (IllegalStateException e) {
        // Catches "call already closed" if another callback or transport closed the stream earlier
    }
}

public void onCompleted() {
    if (closed) {
        return;
    }
    closed = true;
    try {
        responseObserver.onCompleted();
    } catch (IllegalStateException e) {
        // Catches "call already closed" if another callback or transport closed the stream earlier
    }
}

}

5.1. Multi-callback coordination: Does setting closed = true on the first terminal call (responseObserver.onError() or responseObserver.onCompleted()) prevent all normal double-close races between server error, client half-close in requestObserver.onCompleted(), and client cancellation in requestObserver.onError()?

5.2. Memory visibility (JMM): In this synchronous model where all callbacks are dispatched via JumpToApplicationThreadServerStreamListener's per-call SerializingExecutor, does the executor's volatile runState update and task handoff establish a strict happens-before relationship across worker threads, making a plain boolean closed fully safe without volatile overhead on hot streaming paths passing hundreds of messages through responseObserver.onNext()?

5.3. Suppressing vs. Rethrowing IllegalStateException:

5.3.1. On terminal calls (responseObserver.onError() / responseObserver.onCompleted()): when a duplicate close throws IllegalStateException: call already closed, is catching and suppressing it the recommended practice to prevent SerializingExecutor from logging alarming SEVERE: Exception while executing runnable errors on an already-terminated stream?

5.3.2. On responseObserver.onNext(): when an earlier silent write failure causes responseObserver.onNext() to throw IllegalStateException: call is closed, is catching and suppressing it (setting closed = true) recommended to prevent cascading false-positive error logs in production, or should it be rethrown?

5.3.3. Can you confirm that catching and suppressing these IllegalStateExceptions guarantees zero transport or memory leak, because the stream was already closed and deallocated on Netty?

6. Comprehensive Best Practice Confirmation
From a production best-practice perspective, could you confirm whether combining the patterns above represents the idiomatic, recommended architecture for bidirectional streaming:

6.1. All interceptors wrap custom logic in try / catch (Exception) and always execute delegate calls (super) in finally.

6.2. Interceptors never proactively invoke ServerCall.close() on established active streams, relying strictly on cooperative Context.CancellableContext cancellation for any mid-stream limits.

6.3. The application uses the OutboundResponse wrapper to coordinate terminal calls across independent callbacks (requestObserver.onNext(), requestObserver.onError(), requestObserver.onCompleted()), ensuring the first terminal call (responseObserver.onError() or responseObserver.onCompleted()) wins and subsequent calls short-circuit cleanly.

6.4. OutboundResponse catches IllegalStateException: call is closed on responseObserver.onNext() (caused by silent write failures in sendMessageInternal) and flips closed = true, preventing secondary IllegalStateException: call already closed exceptions from cascading across subsequent callbacks.

6.5. Unhandled Exception Lifecycle Behavior (Understanding): When an unhandled exception escapes requestObserver callbacks (requestObserver.onNext(), requestObserver.onError(), or requestObserver.onCompleted()), gRPC core (JumpToApplicationThreadServerStreamListener) catches it and closes the transport stream directly via ServerStream.close(Status.UNKNOWN). Is our understanding correct that ServerCall.close() is completely bypassed, meaning custom interceptor close() logic never runs, and the client receives a generic Status.UNKNOWN?

6.6. Safety of Leaving Unexpected Exceptions Unhandled: From a stability and resource perspective, is letting unexpected exceptions escape unhandled to gRPC core completely fine? (Specifically: attempting to catch and close them generically inside an interceptor is problematic because interceptors lack visibility into responseObserver's internal state and risk double-close races, whereas gRPC core's fallback guarantees complete transport stream teardown, thread survival, and zero resource leaks)?

6.7. Does this overall design represent the recommended production-grade architecture to guarantee clean lifecycle management, zero unhandled IllegalStateException crashes, and zero resource leaks across all normal and error scenarios?

6.8. Applicability to Unary RPCs: For unary RPCs (void method(Request, StreamObserver)), is applying these exact same practices (interceptor exception isolation, finally delegation, admission control before startCall, and wrapping responseObserver in OutboundResponse so that an onError() call during request processing prevents a subsequent onCompleted() from throwing IllegalStateException: call already closed) equally safe, recommended, and guaranteed to provide zero resource leaks?

Thanks,
Aditya

Adding one more point to the above from a best-practices point of view, please confirm the following too.

7. Necessity of setOnCancelHandler in Synchronous Request-Response Flow:
- In a model where responseObserver.onNext() is called strictly and synchronously inside requestObserver.onNext() (and never called from background threads):
  - 7.1. Because client cancellation immediately marks the transport stream cancelled, gRPC core (ServerCallImpl.ServerStreamListenerImpl) automatically drops all subsequent incoming data frames and never invokes requestObserver.onNext() again after cancellation.
  - 7.2. Since requestObserver.onNext() is never invoked after cancellation, responseObserver.onNext() is also never called after cancellation.
  - 7.3. Therefore, does registering ServerCallStreamObserver.setOnCancelHandler() provide any functional or exception-prevention value in a synchronous request-response flow (given that its only effect is suppressing StatusRuntimeException: CANCELLED on responseObserver.onNext()), or is it completely unnecessary and safe to omit?

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

kannanjgithub commented on Oct 23, 2026

adityapallapolu-dot commented on Oct 27, 2026

adityapallapolu-dot commented on Oct 28, 2026

adityapallapolu-dot commented on Oct 28, 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)