Definition
- ServerInterceptor – gRPC's mechanism for cross-cutting concerns; implements
interceptCall(ServerCall, Metadata, ServerCallHandler). - Logging interceptor – wraps every call to log method name, latency, and final status code uniformly.
- Rate-limiter interceptor – counts calls per client with a token bucket and closes excess traffic with
Status.RESOURCE_EXHAUSTED. - Bidirectional streaming – declared with
streamon both request and response in the.proto; both sides exchange messages over one HTTP/2 stream. - StreamObserver – the callback interface for sending and receiving messages in streaming RPCs.
Picture National Bank of India's trading floor. The logging interceptor is the CCTV that records every order with a timestamp and outcome, so nobody has to add cameras to each desk. The rate-limiter interceptor is the doorman who allows only so many orders per second from one trader so a greedy client cannot flood the gateway. The bidirectional stream is the open phone line between a trader and the live price desk: the trader keeps naming stocks, and the desk keeps shouting back new prices, both talking at once until either hangs up.
Understanding the Concepts
How does a gRPC ServerInterceptor work?
A ServerInterceptor implements interceptCall, where you can wrap the ServerCall and Metadata before delegating to the next handler. This is gRPC's hook for cross-cutting concerns.
package com.bank.grpc;
import io.grpc.*;
public class LoggingInterceptor implements ServerInterceptor {
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
ServerCall<ReqT, RespT> call,
Metadata headers,
ServerCallHandler<ReqT, RespT> next) {
long startNanos = System.nanoTime();
String method = call.getMethodDescriptor().getFullMethodName();
// Wrap the ServerCall to observe the status at close
ServerCall<ReqT, RespT> wrappedCall =
new ForwardingServerCall.SimpleForwardingServerCall<>(call) {
@Override
public void close(Status status, Metadata trailers) {
long elapsedMs = (System.nanoTime() - startNanos) / 1_000_000;
System.out.println("[gRPC] " + method
+ " status=" + status.getCode()
+ " latency=" + elapsedMs + "ms");
super.close(status, trailers);
}
};
return next.startCall(wrappedCall, headers);
}
}
[gRPC] com.bank.TradingService/GetBalance status=OK latency=12ms
[gRPC] com.bank.TradingService/Transfer status=INVALID_ARGUMENT latency=3ms
How do you implement a rate-limiter interceptor?
Keep a counter or token bucket keyed by client identity. Reject excess traffic with RESOURCE_EXHAUSTED.
package com.bank.grpc;
import com.google.common.util.concurrent.RateLimiter;
import io.grpc.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class RateLimiterInterceptor implements ServerInterceptor {
// Per-client rate limiter: 100 requests/second
private final ConcurrentMap<String, RateLimiter> limiters
= new ConcurrentHashMap<>();
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
ServerCall<ReqT, RespT> call,
Metadata headers,
ServerCallHandler<ReqT, RespT> next) {
String clientId = call.getAttributes()
.get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR).toString();
RateLimiter limiter = limiters.computeIfAbsent(
clientId, k -> RateLimiter.create(100.0));
if (!limiter.tryAcquire()) {
// Reject: too many requests
call.close(Status.RESOURCE_EXHAUSTED
.withDescription("Rate limit exceeded for " + clientId),
new Metadata());
return new ServerCall.Listener<>() {}; // empty listener
}
return next.startCall(call, headers);
}
}
How do you register interceptors on a gRPC server?
Interceptors are added to the server builder and execute in order for every incoming call.
package com.bank.grpc;
import io.grpc.Server;
import io.grpc.ServerBuilder;
public class GrpcServerSetup {
public static void main(String[] args) throws Exception {
Server server = ServerBuilder.forPort(9090)
.addService(new TradingServiceImpl())
.intercept(new LoggingInterceptor()) // runs first
.intercept(new RateLimiterInterceptor()) // runs second
.build()
.start();
System.out.println("National Bank of India gRPC server on port 9090");
server.awaitTermination();
}
}
How do you define a bidirectional streaming RPC in a proto file?
Declare stream on both the request and response.
syntax = "proto3";
package com.bank;
service TradingService {
// Bidirectional stream: client sends subscriptions,
// server sends price updates
rpc StreamPrices (stream PriceRequest) returns (stream PriceUpdate);
}
message PriceRequest {
string symbol = 1; // e.g., "RELIANCE"
string action = 2; // "SUBSCRIBE" or "UNSUBSCRIBE"
}
message PriceUpdate {
string symbol = 1;
string price = 2; // BigDecimal as string
int64 timestamp = 3;
}
How do you implement the bidirectional streaming service?
Return a StreamObserver for inbound messages and push updates on the response observer.
package com.bank.grpc;
import io.grpc.stub.StreamObserver;
import java.math.BigDecimal;
import java.util.Set;
import java.util.concurrent.*;
public class TradingServiceImpl
extends TradingServiceGrpc.TradingServiceImplBase {
private final ScheduledExecutorService ticker =
Executors.newScheduledThreadPool(1);
@Override
public StreamObserver<PriceRequest> streamPrices(
StreamObserver<PriceUpdate> responseObserver) {
Set<String> subscriptions = ConcurrentHashMap.newKeySet();
// Ticker pushes prices for subscribed symbols
ScheduledFuture<?> tickerTask = ticker.scheduleAtFixedRate(() -> {
for (String symbol : subscriptions) {
BigDecimal price = new BigDecimal("2450.00")
.add(BigDecimal.valueOf(Math.random() * 10));
responseObserver.onNext(PriceUpdate.newBuilder()
.setSymbol(symbol)
.setPrice(price.toPlainString())
.setTimestamp(System.currentTimeMillis())
.build());
}
}, 0, 500, TimeUnit.MILLISECONDS);
// Inbound observer handles subscribe/unsubscribe
return new StreamObserver<>() {
@Override
public void onNext(PriceRequest request) {
if ("SUBSCRIBE".equals(request.getAction())) {
subscriptions.add(request.getSymbol());
System.out.println("Subscribed: " + request.getSymbol());
} else {
subscriptions.remove(request.getSymbol());
System.out.println("Unsubscribed: " + request.getSymbol());
}
}
@Override
public void onError(Throwable t) {
tickerTask.cancel(true); // stop ticker on error
System.out.println("Client error: " + t.getMessage());
}
@Override
public void onCompleted() {
tickerTask.cancel(true); // stop ticker on client done
responseObserver.onCompleted();
System.out.println("Client stream completed");
}
};
}
}
How do you handle backpressure in bidi streaming?
Check isReady() on the ServerCallStreamObserver before pushing so a slow client does not cause unbounded buffering.
package com.bank.grpc;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
public class BackpressureAwareStream {
public void pushPrices(StreamObserver<PriceUpdate> responseObserver) {
ServerCallStreamObserver<PriceUpdate> serverObserver =
(ServerCallStreamObserver<PriceUpdate>) responseObserver;
serverObserver.setOnReadyHandler(() -> {
while (serverObserver.isReady()) {
// Only push when the client can receive
PriceUpdate update = generatePrice();
serverObserver.onNext(update);
}
});
}
private PriceUpdate generatePrice() {
return PriceUpdate.newBuilder()
.setSymbol("RELIANCE")
.setPrice("2455.50")
.setTimestamp(System.currentTimeMillis())
.build();
}
}
What are common mistakes with gRPC streaming?
Forgetting to cancel the background ticker when the client disconnects leaks threads and memory.
// WRONG: ticker keeps running after client disconnects
@Override
public void onCompleted() {
responseObserver.onCompleted();
// ticker still running! Memory and thread leak.
}
// CORRECT: cancel the ticker
@Override
public void onCompleted() {
tickerTask.cancel(true); // stop the background work
responseObserver.onCompleted();
}
What should a senior engineer know?
- For distributed rate limiting across a fleet, use a sliding window in Redis so the limit is shared.
- Interceptors execute in registration order: add logging before rate limiting so rejected calls are also logged.
- Use
ContextandMetadatato propagate trace IDs and authentication tokens through the interceptor chain. - Monitor the number of active bidi streams as a scaling metric.
// Propagate trace ID through metadata
Metadata.Key<String> TRACE_KEY = Metadata.Key.of(
"x-trace-id", Metadata.ASCII_STRING_MARSHALLER);
String traceId = headers.get(TRACE_KEY);
System.out.println("[gRPC] trace=" + traceId + " method=" + method);
Visual Overview
gRPC Interceptor Chain:
┌────────┐ ┌──────────────┐ ┌──────────────┐ ┌─────────┐
│ Client │──▶│ Logging │──▶│ Rate Limiter │──▶│ Service │
│ │ │ Interceptor │ │ Interceptor │ │ Method │
└────────┘ │ │ │ │ └─────────┘
│ log method, │ │ check bucket │
│ latency, │ │ RESOURCE_ │
│ status │ │ EXHAUSTED if │
└──────────────┘ │ over limit │
└──────────────┘
Bidirectional Streaming:
┌────────┐ subscribe(RELIANCE) ┌────────────┐
│ Trader │──────────────────────▶│ Price │
│(Rajesh)│ subscribe(HDFC) │ Service │
│ │──────────────────────▶│ │
│ │◀──price(RELIANCE)─────│ ticker │
│ │◀──price(HDFC)─────────│ (500ms) │
│ │ unsubscribe(HDFC) │ │
│ │──────────────────────▶│ │
│ │◀──price(RELIANCE)─────│ │
└────────┘ └────────────┘
Key Takeaway
This question covers three gRPC building blocks that fit together in a trading-platform service. The logging interceptor and the rate-limiter interceptor are both ServerInterceptors, which is gRPC's mechanism for cross-cutting concerns. The logging interceptor records start time and method name and wraps the ServerCall so that when it closes it logs the method, the final status code, and the elapsed milliseconds, giving uniform audit logging without touching business methods.
The rate-limiter interceptor keeps a per-client token bucket keyed by the remote address. On each call it tries to acquire a token; if none is available it closes the call with RESOURCE_EXHAUSTED and returns an empty listener so the call never reaches the service. For a distributed fleet, use a sliding window in Redis so the limit is shared across instances.
The live stock prices feature is a bidirectional streaming RPC, declared with stream on both request and response in the proto. The trader keeps sending subscribe and unsubscribe messages, and a scheduled ticker keeps pushing fresh prices for subscribed symbols, both directions running independently on one HTTP/2 stream. The senior details to always mention are backpressure and lifecycle: check isReady on the stream observer before pushing, and cancel the ticker when the client disconnects, because a streaming server that forgets to stop its background work leaks threads and memory quickly.
Remember one sentence: logging and rate-limiter interceptors give uniform cross-cutting protection, while bidirectional streaming lets both sides exchange messages continuously on one HTTP/2 call with backpressure via isReady.
---