gRPC server

The module starts a gRPC server based on grpc-java and connects handlers from the application graph to it. A handler is a BindableService, usually a class that extends a generated ...ImplBase and implements unary or streaming RPC methods.

Kora builds the server on the gRPC OkHttp transport, adds the services and ServerInterceptor implementations found in the graph together with its own telemetry interceptor, manages the server lifecycle, and participates in application readiness checks. If configuration parameters are not enough, the resulting builder can be additionally configured in code through a Configurer component.

For a step-by-step walkthrough before the reference details, see gRPC Server and Advanced gRPC Server.

Dependency

Java Kotlin

Dependency in build.gradle:

implementation "io.koraframework:grpc-server"
implementation "io.grpc:grpc-protobuf:1.83.1"
implementation "javax.annotation:javax.annotation-api:1.3.2"

Module:

@KoraApp
public interface Application extends GrpcServerModule { }

Dependency in build.gradle.kts:

implementation("io.koraframework:grpc-server")
implementation("io.grpc:grpc-protobuf:1.83.1")
implementation("javax.annotation:javax.annotation-api:1.3.2")

Module:

@KoraApp
interface Application : GrpcServerModule

The gRPC runtime that ships with io.koraframework:grpc-server is 1.83.1. Every other io.grpc artifact you add — grpc-protobuf, grpc-services, and anything in test scope — must use that same version, see Testing.

Plugin

The code for the gRPC server is generated with the protobuf gradle plugin.

Java Kotlin

Plugin in build.gradle:

plugins {
    id "com.google.protobuf" version "0.10.0"
}

protobuf {
    protoc { artifact = "com.google.protobuf:protoc:4.35.1" }
    plugins {
        grpc { artifact = "io.grpc:protoc-gen-grpc-java:1.83.1" }
    }
    generateProtoTasks {
        all()*.plugins { grpc {} }
    }
}

sourceSets {
    main {
        java {
            srcDirs "build/generated/source/proto/main/grpc"
            srcDirs "build/generated/source/proto/main/java"
        }
    }
}

Plugin in build.gradle.kts:

import com.google.protobuf.gradle.id

plugins {
    id("com.google.protobuf") version ("0.10.0")
}

protobuf {
    protoc { artifact = "com.google.protobuf:protoc:4.35.1" }
    plugins {
        id("grpc") { artifact = "io.grpc:protoc-gen-grpc-java:1.83.1" }
    }
    generateProtoTasks {
        all().forEach { task -> task.plugins { id("grpc") } }
    }
}

sourceSets.main {
    java.srcDir(layout.buildDirectory.dir("generated/source/proto/main/grpc"))
    java.srcDir(layout.buildDirectory.dir("generated/source/proto/main/java"))
}

The plugin generates Java classes, so in a Kotlin project the generated sources are still registered in the java source set.

Configuration

Only port typically needs to be set; all other parameters have defaults. A minimal configuration that binds a port and enables logging looks like this:

Hocon YAML

grpcServer {
    port = 8090
    telemetry.logging.enabled = true
}

grpcServer:
  port: 8090
  telemetry:
    logging:
      enabled: true

Basic configuration parameters:

Hocon YAML

grpcServer {
    port = 8090 //(1)!
    maxMessageSize = "4MiB" //(2)!
    reflectionEnabled = false //(3)!
}

1. gRPC server port (default: 8090).
2. Maximum incoming message size (default: 4MiB).
3. Enables the gRPC Server Reflection service (default: false).

grpcServer:
  port: 8090 #(1)!
  maxMessageSize: "4MiB" #(2)!
  reflectionEnabled: false #(3)!

1. gRPC server port (default: 8090).
2. Maximum incoming message size (default: 4MiB).
3. Enables the gRPC Server Reflection service (default: false).

Full Configuration

Example of a complete configuration described by GrpcServerConfig:

Hocon YAML

grpcServer {
    port = 8090 //(1)!
    maxMessageSize = "4MiB" //(2)!
    reflectionEnabled = false //(3)!
    shutdownWait = "30s" //(4)!
    maxConnectionAge = "5m" //(5)!
    maxConnectionAgeGrace = "30s" //(6)!
    keepAliveTime = "30s" //(7)!
    keepAliveTimeout = "10s" //(8)!
    telemetry {
        logging {
            enabled = false //(9)!
        }
        metrics {
            enabled = false //(10)!
            slo = [ 1, 10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 90000 ] //(11)!
            tags = { // (12)!
                "key1" = "value1"
                "key2" = "value2"
            }
        }
        tracing {
            enabled = true //(13)!
            attributes = { // (14)!
                "key1" = "value1"
                "key2" = "value2"
            }
        }
    }
}

1. gRPC server port (default: 8090).
2. Maximum size of an incoming message (default: 4MiB). It can be specified as a number of bytes or as 4MiB, 4MB, 1000Kb, and similar values.
3. Enables the gRPC Server Reflection service (default: false).
4. Time to wait for in-flight calls to complete before shutting down the server during graceful shutdown (default: 30s).
5. Maximum connection age after which the connection is gracefully terminated (optional, no default). A random jitter of +/-10% is added to the value.
6. Additional time for graceful connection termination after the maximum connection age is reached (optional, no default). RPC calls that do not finish in time are cancelled so the connection can terminate.
7. Interval between PING frames (optional, no default).
8. Timeout for acknowledging a PING frame (optional, no default). If no acknowledgement is received within this time, the connection is closed.
9. Enables module logging (default: false).
10. Enables module metrics (default: false). Metrics are only reported if a metrics module also provides a MeterRegistry.
11. Configures SLO for the Timer metric (default: io.koraframework.telemetry.common.TelemetryConfig.MetricsConfig#DEFAULT_SLO).
12. Metric tags (default: {}).
13. Enables module tracing (default: true). Spans are only exported if a tracing module also provides a Tracer.
14. Tracing attributes (default: {}).

grpcServer:
  port: 8090 #(1)!
  maxMessageSize: "4MiB" #(2)!
  reflectionEnabled: false #(3)!
  shutdownWait: "30s" #(4)!
  maxConnectionAge: "5m" #(5)!
  maxConnectionAgeGrace: "30s" #(6)!
  keepAliveTime: "30s" #(7)!
  keepAliveTimeout: "10s" #(8)!
  telemetry:
    logging:
      enabled: false #(9)!
    metrics:
      enabled: false #(10)!
      slo: [ 1, 10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 90000 ] #(11)!
      tags: #(12)!
        key1: value1
        key2: value2
    tracing:
      enabled: true #(13)!
      attributes: #(14)!
        key1: value1
        key2: value2

1. gRPC server port (default: 8090).
2. Maximum size of an incoming message (default: 4MiB). It can be specified as a number of bytes or as 4MiB, 4MB, 1000Kb, and similar values.
3. Enables the gRPC Server Reflection service (default: false).
4. Time to wait for in-flight calls to complete before shutting down the server during graceful shutdown (default: 30s).
5. Maximum connection age after which the connection is gracefully terminated (optional, no default). A random jitter of +/-10% is added to the value.
6. Additional time for graceful connection termination after the maximum connection age is reached (optional, no default). RPC calls that do not finish in time are cancelled so the connection can terminate.
7. Interval between PING frames (optional, no default).
8. Timeout for acknowledging a PING frame (optional, no default). If no acknowledgement is received within this time, the connection is closed.
9. Enables module logging (default: false).
10. Enables module metrics (default: false). Metrics are only reported if a metrics module also provides a MeterRegistry.
11. Configures SLO for the Timer metric (default: io.koraframework.telemetry.common.TelemetryConfig.MetricsConfig#DEFAULT_SLO).
12. Metric tags (default: {}).
13. Enables module tracing (default: true). Spans are only exported if a tracing module also provides a Tracer.
14. Tracing attributes (default: {}).

Everything the configuration does not cover is available through configuration in code.

Configuration in code

If configuration parameters are not enough, register a Configurer<ForwardingServerBuilder<?>> component and additionally configure the server builder in code. The component is called last: after the configuration has been applied and after the services, the user-defined ServerInterceptor implementations and the standard interceptor have been added. The builder it returns is the one the server is built from.

Java Kotlin

@Component
public final class MyGrpcServerConfigurer implements Configurer<ForwardingServerBuilder<?>> {

    @Override
    public ForwardingServerBuilder<?> configure(ForwardingServerBuilder<?> builder) {
        builder.maxInboundMetadataSize(16 * 1024); //(1)!
        builder.handshakeTimeout(10, TimeUnit.SECONDS);

        if (builder instanceof OkHttpServerBuilder okHttpBuilder) { //(2)!
            okHttpBuilder.permitKeepAliveWithoutCalls(true);
            okHttpBuilder.maxConcurrentCallsPerConnection(200);
        }

        return builder; //(3)!
    }
}

1. Options declared by io.grpc.ServerBuilder are available on the forwarding builder directly
2. Transport-specific options require the concrete builder: Kora runs the server on the gRPC OkHttp transport, so it is an io.grpc.okhttp.OkHttpServerBuilder
3. gRPC builders are mutable and return themselves, so the same instance is handed back

@Component
class MyGrpcServerConfigurer : Configurer<ForwardingServerBuilder<*>> {

    override fun configure(builder: ForwardingServerBuilder<*>): ForwardingServerBuilder<*> {
        builder.maxInboundMetadataSize(16 * 1024) //(1)!
        builder.handshakeTimeout(10, TimeUnit.SECONDS)

        if (builder is OkHttpServerBuilder) { //(2)!
            builder.permitKeepAliveWithoutCalls(true)
            builder.maxConcurrentCallsPerConnection(200)
        }

        return builder //(3)!
    }
}

1. Options declared by io.grpc.ServerBuilder are available on the forwarding builder directly
2. Transport-specific options require the concrete builder: Kora runs the server on the gRPC OkHttp transport, so it is an io.grpc.okhttp.OkHttpServerBuilder
3. gRPC builders are mutable and return themselves, so the same instance is handed back

Module metrics are described in the Metrics Reference section.

Transport security

By default the server accepts plaintext connections: when the graph contains no io.grpc.ServerCredentials, Kora uses InsecureServerCredentials. To terminate TLS in the server itself, provide ServerCredentials as a component — for example with io.grpc.TlsServerCredentials:

Java Kotlin

@KoraApp
public interface Application extends GrpcServerModule {

    default ServerCredentials grpcServerCredentials() throws IOException { //(1)!
        return TlsServerCredentials.create(
            new File("/etc/certs/server.crt"), //(2)!
            new File("/etc/certs/server.key"));
    }
}

1. A factory method of the application graph: the credentials are picked up when the gRPC server builder is created
2. A PEM-encoded certificate chain and an unencrypted PKCS#8 private key

@KoraApp
interface Application : GrpcServerModule {

    fun grpcServerCredentials(): ServerCredentials { //(1)!
        return TlsServerCredentials.create(
            File("/etc/certs/server.crt"), //(2)!
            File("/etc/certs/server.key"))
    }
}

1. A factory method of the application graph: the credentials are picked up when the gRPC server builder is created
2. A PEM-encoded certificate chain and an unencrypted PKCS#8 private key

For mutual TLS and custom trust stores build the credentials with TlsServerCredentials.newBuilder() instead.

Handlers

A handler is a class that extends the generated ...ImplBase and is registered in the application graph with the @Component annotation. The ...ImplBase class is produced from the proto contract by the protobuf gradle plugin; you override its RPC methods to implement server behavior. Ordinary Kora components such as services and repositories can be injected into a handler through its constructor.

Consider a proto contract with a single unary method:

src/main/proto/message.proto

syntax = "proto3";

package io.koraframework.generated.grpc;

service UserService {
  rpc createUser(RequestEvent) returns (ResponseEvent) {} //(1)!
}

message RequestEvent {
  string name = 1;
  string code = 2;
}

message ResponseEvent {
  bytes id = 1;
}

1. A unary RPC: one request message produces one response message.

The plugin generates UserServiceGrpc.UserServiceImplBase, and the handler overrides the generated method. The generated method receives the request message and a StreamObserver that is used to send responses back to the client:

Java Kotlin

@Component
public final class UserService extends UserServiceGrpc.UserServiceImplBase {

    @Override
    public void createUser(Message.RequestEvent request, StreamObserver<Message.ResponseEvent> responseObserver) { //(1)!
        var response = Message.ResponseEvent.newBuilder()
            .setId(ByteString.copyFromUtf8(UUID.randomUUID().toString()))
            .build();

        responseObserver.onNext(response); //(2)!
        responseObserver.onCompleted(); //(3)!
    }
}

1. The generated method receives the request message and a StreamObserver for sending the response
2. Sends a single response message to the client
3. Signals that the call is complete; for a unary method it is called exactly once, after a single onNext

@Component
class UserService : UserServiceGrpc.UserServiceImplBase() {

    override fun createUser(request: Message.RequestEvent, responseObserver: StreamObserver<Message.ResponseEvent>) { //(1)!
        val response = Message.ResponseEvent.newBuilder()
            .setId(ByteString.copyFromUtf8(UUID.randomUUID().toString()))
            .build()

        responseObserver.onNext(response) //(2)!
        responseObserver.onCompleted() //(3)!
    }
}

1. The generated method receives the request message and a StreamObserver for sending the response
2. Sends a single response message to the client
3. Signals that the call is complete; for a unary method it is called exactly once, after a single onNext

Server streaming

For a server-streaming RPC (returns (stream ...) in the proto), the client sends one request and the server sends back many messages. Call onNext for each message, then onCompleted once at the end:

Java Kotlin

@Override
public void getAllUsers(Message.RequestEvent request, StreamObserver<Message.ResponseEvent> responseObserver) {
    for (var user : userService.findAll()) {
        responseObserver.onNext(toResponse(user)); //(1)!
    }
    responseObserver.onCompleted(); //(2)!
}

1. Sends one of several response messages
2. Completes the response stream after the last message

override fun getAllUsers(request: Message.RequestEvent, responseObserver: StreamObserver<Message.ResponseEvent>) {
    userService.findAll().forEach { responseObserver.onNext(toResponse(it)) } //(1)!
    responseObserver.onCompleted() //(2)!
}

1. Sends one of several response messages
2. Completes the response stream after the last message

Client streaming

For a client-streaming RPC (rpc method(stream ...)), the client sends many messages and the server answers once at the end. The generated method returns a StreamObserver that receives the incoming request messages; the final response is produced from onCompleted:

Java Kotlin

@Override
public StreamObserver<Message.RequestEvent> createUsers(StreamObserver<Message.ResponseEvent> responseObserver) {
    return new StreamObserver<>() {
        private final List<Message.RequestEvent> received = new ArrayList<>();

        @Override
        public void onNext(Message.RequestEvent value) {
            received.add(value); //(1)!
        }

        @Override
        public void onError(Throwable t) {
            responseObserver.onError(t); //(2)!
        }

        @Override
        public void onCompleted() {
            responseObserver.onNext(aggregate(received)); //(3)!
            responseObserver.onCompleted();
        }
    };
}

1. Collects each incoming request message
2. Propagates a client-side stream error
3. Produces the single aggregated response once the client has finished sending

override fun createUsers(responseObserver: StreamObserver<Message.ResponseEvent>): StreamObserver<Message.RequestEvent> {
    return object : StreamObserver<Message.RequestEvent> {
        private val received = mutableListOf<Message.RequestEvent>()

        override fun onNext(value: Message.RequestEvent) {
            received += value //(1)!
        }

        override fun onError(t: Throwable) {
            responseObserver.onError(t) //(2)!
        }

        override fun onCompleted() {
            responseObserver.onNext(aggregate(received)) //(3)!
            responseObserver.onCompleted()
        }
    }
}

1. Collects each incoming request message
2. Propagates a client-side stream error
3. Produces the single aggregated response once the client has finished sending

Bidirectional streaming

For a bidirectional-streaming RPC (rpc method(stream ...) returns (stream ...)), both sides exchange many messages on the same call. The method returns a StreamObserver for the incoming requests and can send responses at any time through responseObserver:

Java Kotlin

@Override
public StreamObserver<Message.RequestEvent> updateUsers(StreamObserver<Message.ResponseEvent> responseObserver) {
    return new StreamObserver<>() {
        @Override
        public void onNext(Message.RequestEvent value) {
            responseObserver.onNext(process(value)); //(1)!
        }

        @Override
        public void onError(Throwable t) {
            responseObserver.onError(t);
        }

        @Override
        public void onCompleted() {
            responseObserver.onCompleted(); //(2)!
        }
    };
}

1. Responds to each incoming message as it arrives
2. Completes the response stream when the client stops sending

override fun updateUsers(responseObserver: StreamObserver<Message.ResponseEvent>): StreamObserver<Message.RequestEvent> {
    return object : StreamObserver<Message.RequestEvent> {
        override fun onNext(value: Message.RequestEvent) {
            responseObserver.onNext(process(value)) //(1)!
        }

        override fun onError(t: Throwable) {
            responseObserver.onError(t)
        }

        override fun onCompleted() {
            responseObserver.onCompleted() //(2)!
        }
    }
}

1. Responds to each incoming message as it arrives
2. Completes the response stream when the client stops sending

Error handling

Description: gRPC represents call errors with an io.grpc.Status code and an optional description rather than with HTTP response codes. To fail a call, complete the response observer with responseObserver.onError(status.asRuntimeException()), or throw a StatusRuntimeException from the handler. The auto-registered TelemetryInterceptor observes the terminal Status when the call is closed and records logging, metrics, and tracing accordingly.

Causes: choose the Status code that matches the failure — for example Status.NOT_FOUND for a missing entity, Status.INVALID_ARGUMENT for invalid input, Status.UNAUTHENTICATED or Status.PERMISSION_DENIED for authorization failures, and Status.INTERNAL for unexpected server errors.

Recommendations:

• Attach a human-readable message with withDescription(...) and keep the original exception with withCause(...) so telemetry can record it.
• Complete a call exactly once: never call onError after onCompleted, and never call either twice.
• Do not leak internal exception details to clients; map them to an appropriate Status first.

Handling example: a unary handler that returns NOT_FOUND when an entity is missing and maps unexpected failures to INTERNAL:

Java Kotlin

@Override
public void getUser(Message.RequestEvent request, StreamObserver<Message.ResponseEvent> responseObserver) {
    try {
        var user = userService.getUser(request.getName())
            .orElseThrow(() -> Status.NOT_FOUND
                .withDescription("User not found: " + request.getName())
                .asRuntimeException()); //(1)!
        responseObserver.onNext(toResponse(user));
        responseObserver.onCompleted();
    } catch (StatusRuntimeException e) {
        responseObserver.onError(e); //(2)!
    } catch (Exception e) {
        responseObserver.onError(Status.INTERNAL
            .withDescription("Failed to get user")
            .withCause(e) //(3)!
            .asRuntimeException());
    }
}

1. Builds a NOT_FOUND error with a description
2. Forwards an already-mapped Status error to the client
3. Keeps the original exception as the cause so telemetry can record it

override fun getUser(request: Message.RequestEvent, responseObserver: StreamObserver<Message.ResponseEvent>) {
    try {
        val user = userService.getUser(request.name)
            ?: throw Status.NOT_FOUND
                .withDescription("User not found: ${request.name}")
                .asRuntimeException() //(1)!
        responseObserver.onNext(toResponse(user))
        responseObserver.onCompleted()
    } catch (e: StatusRuntimeException) {
        responseObserver.onError(e) //(2)!
    } catch (e: Exception) {
        responseObserver.onError(
            Status.INTERNAL
                .withDescription("Failed to get user")
                .withCause(e) //(3)!
                .asRuntimeException()
        )
    }
}

1. Builds a NOT_FOUND error with a description
2. Forwards an already-mapped Status error to the client
3. Keeps the original exception as the cause so telemetry can record it

Signatures

The shape of a handler method is fixed by the proto contract and the generated ...ImplBase:

Java Kotlin

By Req and Resp we mean the generated request and response message types.

• Unary: void myMethod(Req request, StreamObserver<Resp> responseObserver)
• Server streaming: void myMethod(Req request, StreamObserver<Resp> responseObserver) (multiple onNext, one onCompleted)
• Client streaming: StreamObserver<Req> myMethod(StreamObserver<Resp> responseObserver)
• Bidirectional streaming: StreamObserver<Req> myMethod(StreamObserver<Resp> responseObserver)

The generated method returns void (or the request StreamObserver), so results are delivered through the StreamObserver callbacks rather than through the return value.

By Req and Resp we mean the generated request and response message types.

• Unary: myMethod(request: Req, responseObserver: StreamObserver<Resp>)
• Server streaming: myMethod(request: Req, responseObserver: StreamObserver<Resp>) (multiple onNext, one onCompleted)
• Client streaming: myMethod(responseObserver: StreamObserver<Resp>): StreamObserver<Req>
• Bidirectional streaming: myMethod(responseObserver: StreamObserver<Resp>): StreamObserver<Req>

The generated method returns Unit (or the request StreamObserver), so results are delivered through the StreamObserver callbacks rather than through the return value.

Handlers are blocking: a handler method may call a database, an HTTP client or a gRPC client directly. There are no asynchronous, reactive or suspend handler signatures — the server runs handlers on virtual threads, see Execution model.

Execution model

Every client connection gets a dedicated single-threaded executor backed by a virtual thread named grpc-<remote address>. The executor is created when the transport becomes ready and shut down when the transport terminates.

• All interceptor and handler callbacks of the calls arriving on one connection run on that one virtual thread, one at a time and in arrival order.
• Blocking inside a handler is safe — the carrier thread is released — but it delays the other calls of the same connection. A client that needs concurrent calls should open several connections.
• Kora binds its MDC and the OpenTelemetry context to that thread for the duration of each callback, so logging context is available inside the handler.

Interceptors

An io.grpc.ServerInterceptor processes a call before it is passed to a gRPC service. Interceptors are suitable for cross-cutting logic: logging, authorization, tracing, working with Metadata, and error mapping.

Unlike the HTTP server, the gRPC server module has no @GrpcService or @InterceptWith annotation: every ServerInterceptor registered as a @Component is applied globally to all services on the server. To limit an interceptor to a single service or method, inspect the call at runtime — see Scoping and authorization.

Default

When the server starts, Kora adds one standard interceptor:

• TelemetryInterceptor — opens a GrpcServerObservation for every call, binds the current observation and the OpenTelemetry context around the call, and records logging, metrics and tracing when the call is closed, depending on the connected modules and the grpcServer.telemetry settings

User-defined ServerInterceptor components from the application graph are added to the builder before the standard interceptor. For full builder configuration, use configuration in code.

Interceptor components are read through the refreshable graph, so an interceptor that is rebuilt on a configuration refresh is picked up without restarting the server.

Execution order

gRPC invokes interceptors in the reverse order of registration, so the last interceptor added runs first (outermost). Because Kora registers user interceptors first and the standard interceptor last, an incoming call is processed in this order:

TelemetryInterceptor -> user interceptors -> handler

Consequences of this order:

• TelemetryInterceptor wraps your interceptors and the handler, so it observes the final Status — including errors thrown by your interceptors or reported through the response observer.
• The current observation and the OpenTelemetry context are established around your interceptors and the handler, so they are available inside the handler's listener callbacks.
• When several user interceptors exist, they run in the reverse of their graph registration order; do not rely on a specific order between them for correctness.

Custom

To add a custom interceptor, create a ServerInterceptor implementation with the @Component annotation:

Java Kotlin

@Component
public final class LoggingServerInterceptor implements ServerInterceptor {

    private static final Logger logger = LoggerFactory.getLogger(LoggingServerInterceptor.class);

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
                                                                 Metadata headers,
                                                                 ServerCallHandler<ReqT, RespT> next) {
        logger.info("Incoming gRPC call: {}", call.getMethodDescriptor().getFullMethodName()); //(1)!

        return next.startCall(call, headers); //(2)!
    }
}

1. getFullMethodName() returns service/method for the intercepted call
2. Passes the call on; if you return without calling startCall, you must close the call yourself

@Component
class LoggingServerInterceptor : ServerInterceptor {

    private val logger = LoggerFactory.getLogger(LoggingServerInterceptor::class.java)

    override fun <ReqT : Any, RespT : Any> interceptCall(
        call: ServerCall<ReqT, RespT>,
        headers: Metadata,
        next: ServerCallHandler<ReqT, RespT>
    ): ServerCall.Listener<ReqT> {
        logger.info("Incoming gRPC call: {}", call.methodDescriptor.fullMethodName) //(1)!

        return next.startCall(call, headers) //(2)!
    }
}

1. fullMethodName returns service/method for the intercepted call
2. Passes the call on; if you return without calling startCall, you must close the call yourself

Scoping and authorization

Because an interceptor is global, scope it to a specific service or method by inspecting call.getMethodDescriptor(): getServiceName() returns the service name (the generated constant ...Grpc.SERVICE_NAME), and getFullMethodName() returns service/method.

Request headers arrive as Metadata. Read a header with a Metadata.Key, and reject a call by closing it with a Status and returning an empty listener so the handler is never invoked. The example below applies API-key authorization to a single service only:

Java Kotlin

@ConfigSource("auth.apiKey")
public interface ApiKeyConfig {

    String value();
}

@Component
public final class ApiKeyServerInterceptor implements ServerInterceptor {

    private static final Metadata.Key<String> AUTHORIZATION =
        Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER); //(1)!

    private final ApiKeyConfig config;

    public ApiKeyServerInterceptor(ApiKeyConfig config) {
        this.config = config;
    }

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
                                                                 Metadata headers,
                                                                 ServerCallHandler<ReqT, RespT> next) {
        if (!UserServiceGrpc.SERVICE_NAME.equals(call.getMethodDescriptor().getServiceName())) { //(2)!
            return next.startCall(call, headers);
        }

        var apiKey = headers.get(AUTHORIZATION); //(3)!
        if (!config.value().equals(apiKey)) {
            call.close(Status.UNAUTHENTICATED.withDescription("Invalid API key"), new Metadata()); //(4)!
            return new ServerCall.Listener<>() {}; //(5)!
        }

        return next.startCall(call, headers);
    }
}

1. Metadata.Key for reading the authorization header as an ASCII string
2. Applies the interceptor only to UserService; other services pass through untouched
3. Reads the header value from the request Metadata
4. Rejects the call with an UNAUTHENTICATED status
5. Returns an empty listener so the handler is never called

@ConfigSource("auth.apiKey")
interface ApiKeyConfig {

    fun value(): String
}

@Component
class ApiKeyServerInterceptor(private val config: ApiKeyConfig) : ServerInterceptor {

    override fun <ReqT : Any, RespT : Any> interceptCall(
        call: ServerCall<ReqT, RespT>,
        headers: Metadata,
        next: ServerCallHandler<ReqT, RespT>
    ): ServerCall.Listener<ReqT> {
        if (UserServiceGrpc.SERVICE_NAME != call.methodDescriptor.serviceName) { //(2)!
            return next.startCall(call, headers)
        }

        val apiKey = headers.get(AUTHORIZATION) //(3)!
        if (config.value() != apiKey) {
            call.close(Status.UNAUTHENTICATED.withDescription("Invalid API key"), Metadata()) //(4)!
            return object : ServerCall.Listener<ReqT>() {} //(5)!
        }

        return next.startCall(call, headers)
    }

    companion object {
        private val AUTHORIZATION: Metadata.Key<String> =
            Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER) //(1)!
    }
}

1. Metadata.Key for reading the authorization header as an ASCII string
2. Applies the interceptor only to UserService; other services pass through untouched
3. Reads the header value from the request Metadata
4. Rejects the call with an UNAUTHENTICATED status
5. Returns an empty listener so the handler is never called

Lifecycle and readiness

The server is managed by the GrpcServer component, which is created as a @Root component and follows the application lifecycle:

• On startup it builds and starts the server on the configured port. If the port is already in use, startup fails with gRPC server failed to start on port '8090': port is already in use; stop the other process or configure a different port.
• On shutdown it performs a graceful shutdown: it stops accepting new calls and waits up to shutdownWait for in-flight calls to finish, then forcibly terminates any remaining calls.

GrpcServer also implements a readiness probe: the server reports not ready while it is starting up or shutting down, and ready only while it is running. In a Kubernetes deployment this lets the readiness probe reflect the real server state and drain traffic during graceful shutdown.

Reflection

gRPC Server Reflection is supported and provides information about available gRPC services on the server. Reflection helps clients and tools build RPC requests at runtime without precompiled service information. For example, it is used by gRPC CLI, which can inspect server proto descriptions and send test RPC calls. gRPC Server Reflection is supported only for proto-based services.

You can learn more about gRPC Server Reflection in the grpc-java guide.

Dependency

You must additionally add the gRPC Server Reflection dependency.

Java Kotlin

Dependency in build.gradle:

implementation "io.grpc:grpc-services:1.83.1"

Dependency in build.gradle.kts:

implementation("io.grpc:grpc-services:1.83.1")

Configuration

You must also enable the gRPC Server Reflection service in the configuration. Kora adds it to the server only if the application has the io.grpc.protobuf.services.ProtoReflectionServiceV1 class, so configuration alone is not enough without the dependency.

Hocon YAML

grpcServer {
    reflectionEnabled = false //(1)!
}

1. Enables the gRPC Server Reflection service (default: false).

grpcServer:
  reflectionEnabled: false #(1)!

1. Enables the gRPC Server Reflection service (default: false).

Usage

With reflection enabled, tools such as grpcurl can discover services and send RPC calls without a precompiled client. For a server listening on port 8090:

grpcurl -plaintext localhost:8090 list #(1)!
grpcurl -plaintext localhost:8090 describe io.koraframework.generated.grpc.UserService #(2)!
grpcurl -plaintext -d '{"name": "Bob", "code": "123"}' \
    localhost:8090 io.koraframework.generated.grpc.UserService/createUser #(3)!

1. Lists the services exposed by the server
2. Describes a service and its methods
3. Sends a unary RPC; -plaintext is used because the example server has no TLS

Telemetry

Server observability is driven by the TelemetryInterceptor through the GrpcServerTelemetry facade and is configured under grpcServer.telemetry. Extension points are located in io.koraframework.grpc.server.telemetry.

For each gRPC call a GrpcServerObservation is created; it collects headers, sent and received messages, the terminal Status and the error, and is ended when the call is closed. The default GrpcServerTelemetryFactory is registered as a @DefaultComponent, so it can be replaced entirely; alternatively the individual parts can be overridden by registering a DefaultGrpcServerLoggerFactory, DefaultGrpcServerMetricsFactory or DefaultGrpcServerBodyConverter subclass as a component. When logging, metrics and tracing are all disabled the factory returns a no-op telemetry, and calls carry no observability overhead at all.

The server reports itself under the name kora-grpc in logs, metrics and spans.

Logging

Call logging is enabled with grpcServer.telemetry.logging.enabled and is written by two loggers:

• io.koraframework.grpc.server.GrpcServer.request — GrpcCall received, with the grpcRequest structured field
• io.koraframework.grpc.server.GrpcServer.response — GrpcCall responded, with the grpcResponse structured field

The structured fields carry serverName, serverPort, serviceName and operation (service/method); the response also carries processingTime in milliseconds, the Status code in status and, for a failed call, exceptionType. A call that ends with an error is logged at WARN with the exception attached, a successful one at INFO.

The logger level adds detail on top of that: DEBUG on the request logger adds the request Metadata in headers, and TRACE adds the message body, rendered by DefaultGrpcServerBodyConverter.

Hocon YAML

logging.levels {
    "io.koraframework.grpc.server.GrpcServer.request" = "DEBUG"
    "io.koraframework.grpc.server.GrpcServer.response" = "TRACE"
}

logging:
  levels:
    "io.koraframework.grpc.server.GrpcServer.request": "DEBUG"
    "io.koraframework.grpc.server.GrpcServer.response": "TRACE"

Metrics

Metrics require grpcServer.telemetry.metrics.enabled and a MeterRegistry supplied by a metrics module. The module reports a single rpc.server.duration timer, with the buckets from grpcServer.telemetry.metrics.slo and the tags server.name, server.port, rpc.system (always grpc), rpc.service, rpc.method and rpc.grpc.status_code, plus everything declared in grpcServer.telemetry.metrics.tags.

Metrics are described in the Metrics Reference section.

Tracing

Tracing requires grpcServer.telemetry.tracing.enabled and a Tracer supplied by a tracing module. A SERVER span named <service>/<method> is created for every call; its parent is extracted from the request Metadata using the W3C Trace Context propagator, so a trace started by the caller continues on the server.

The span carries the server.port, server.name, rpc.system, rpc.service, rpc.method and network.peer.address attributes, plus everything declared in grpcServer.telemetry.tracing.attributes; on close rpc.grpc.status_code is added. Each sent and received message adds an rpc.message event with the rpc.message.type attribute. A non-OK Status or an exception marks the span status as ERROR.

Testing

The gRPC server binds a real port, so a @KoraAppTest starts it and the test talks to it through an ordinary ManagedChannel:

Java Kotlin

@KoraAppTest(Application.class)
class UserServiceTests {

    @Test
    void createUser() {
        var channel = ManagedChannelBuilder.forAddress("localhost", 8090) //(1)!
            .usePlaintext()
            .build();

        try {
            var stub = UserServiceGrpc.newBlockingStub(channel); //(2)!
            var response = stub.createUser(Message.RequestEvent.newBuilder()
                .setName("Bob")
                .setCode("123")
                .build());

            assertFalse(response.getId().isEmpty());
        } finally {
            channel.shutdownNow();
        }
    }
}

1. The port the server was started on, that is grpcServer.port from the test configuration
2. A blocking stub generated from the proto contract

@KoraAppTest(Application::class)
class UserServiceTests {

    @Test
    fun createUser() {
        val channel = ManagedChannelBuilder.forAddress("localhost", 8090) //(1)!
            .usePlaintext()
            .build()

        try {
            val stub = UserServiceGrpc.newBlockingStub(channel) //(2)!
            val response = stub.createUser(
                Message.RequestEvent.newBuilder()
                    .setName("Bob")
                    .setCode("123")
                    .build()
            )

            assertFalse(response.id.isEmpty)
        } finally {
            channel.shutdownNow()
        }
    }
}

1. The port the server was started on, that is grpcServer.port from the test configuration
2. A blocking stub generated from the proto contract

Version alignment: the client side of a test needs a gRPC transport on the test classpath, and its version must match the gRPC runtime that comes with io.koraframework:grpc-server — 1.83.1. A pinned older version compiles fine and fails only at runtime with AbstractMethodError: ... does not define or inherit an implementation of the resolved method 'buildClientTransportServers(List, MetricRecorder)'.

Java Kotlin

Dependency in build.gradle:

testImplementation "io.koraframework:test-junit5"
testImplementation "io.grpc:grpc-netty:1.83.1"

Dependency in build.gradle.kts:

testImplementation("io.koraframework:test-junit5")
testImplementation("io.grpc:grpc-netty:1.83.1")