gRPC trong Spring Boot

syntax = "proto3";
 
package user;
 
option java_multiple_files = true;
option java_package = "com.tungdadev.grpc.user";
 
// 1. Service definition
service UserService {
    // Unary RPC
    rpc GetUser (GetUserRequest) returns (UserResponse);
    // Server streaming
    rpc ListUsers (ListUsersRequest) returns (stream UserResponse);
    // Client streaming
    rpc UploadUsers (stream CreateUserRequest) returns (UploadSummary);
    // Bidirectional streaming
    rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}
 
// 2. Messages definition
message GetUserRequest {
    string user_id = 1;
}
 
message UserResponse {
    string id = 1;
    string name = 2;
    string email = 3;
    int32 age = 4;
    repeated string roles = 5;        // Tương đương List trong Java
    map<string, string> metadata = 6; // Tương đương Map
    UserStatus status = 7;
    optional string phone = 8;        // Nullable field (Proto3)
}
 
enum UserStatus {
    USER_STATUS_UNSPECIFIED = 0;      // Bắt buộc giá trị mặc định là 0
    USER_STATUS_ACTIVE = 1;
    USER_STATUS_INACTIVE = 2;
    USER_STATUS_BANNED = 3;
}

// Server implementation
@GrpcService
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {
    @Override
    public void getUser(GetUserRequest request, StreamObserver<UserResponse> responseObserver) {
        UserResponse user = UserResponse.newBuilder()
            .setId(request.getUserId())
            .setName("John Doe")
            .build();
 
        responseObserver.onNext(user);
        responseObserver.onCompleted(); // Đóng stream
    }
}

// Server
@Override
public void listUsers(ListUsersRequest request, StreamObserver<UserResponse> responseObserver) {
   List<User> users = userRepository.findAll(request.getFilter());
 
   for (User user : users) {
       responseObserver.onNext(toProto(user)); // Gửi từng chunk
   }
   responseObserver.onCompleted();
}
 
// Client
Iterator<UserResponse> users = userStub.listUsers(request);
while (users.hasNext()) {
   UserResponse user = users.next();
   // process each user
}

// Server
@Override
public StreamObserver<CreateUserRequest> uploadUsers(StreamObserver<UploadSummary> responseObserver) {
   return new StreamObserver<>() {
       int received = 0, created = 0, failed = 0;
 
       @Override
       public void onNext(CreateUserRequest request) {
           received++;
           try {
               userRepository.save(toEntity(request));
               created++;
           } catch (Exception e) {
               failed++;
           }
       }
 
       @Override
       public void onCompleted() {
           responseObserver.onNext(UploadSummary.newBuilder()
               .setTotalReceived(received)
               .setTotalCreated(created)
               .setTotalFailed(failed)
               .build());
           responseObserver.onCompleted();
       }
 
       @Override
       public void onError(Throwable t) {
           log.error("Client stream error", t);
       }
   };
}

// Server
@Override
public StreamObserver<ChatMessage> chat(StreamObserver<ChatMessage> responseObserver) {
   return new StreamObserver<>() {
       @Override
       public void onNext(ChatMessage message) {
           // Echo back or broadcast
           ChatMessage reply = ChatMessage.newBuilder()
               .setSenderId("server")
               .setContent("Received: " + message.getContent())
               .setTimestamp(System.currentTimeMillis())
               .build();
           responseObserver.onNext(reply);
       }
 
       @Override
       public void onCompleted() {
           responseObserver.onCompleted();
       }
 
       @Override
       public void onError(Throwable t) {
           log.error("Stream error", t);
       }
   };
}

@Override
public void getUser(GetUserRequest request, StreamObserver<UserResponse> responseObserver) {
   if (request.getUserId().isBlank()) {
       responseObserver.onError(Status.INVALID_ARGUMENT
           .withDescription("user_id must not be empty")
           .asRuntimeException());
       return;
   }
 
   User user = userRepository.findById(request.getUserId()).orElse(null);
   if (user == null) {
       responseObserver.onError(Status.NOT_FOUND
           .withDescription("User not found: " + request.getUserId())
           .asRuntimeException());
       return;
   }
 
   responseObserver.onNext(toProto(user));
   responseObserver.onCompleted();
}

import com.google.protobuf.Any;
import com.google.rpc.BadRequest;
import com.google.rpc.Code;
import io.grpc.protobuf.StatusProto;
 
// Trả error với metadata chi tiết
com.google.rpc.Status status = com.google.rpc.Status.newBuilder()
   .setCode(Code.INVALID_ARGUMENT.getNumber())
   .setMessage("Validation failed")
   .addDetails(Any.pack(BadRequest.newBuilder()
       .addFieldViolations(BadRequest.FieldViolation.newBuilder()
           .setField("email")
           .setDescription("Invalid email format")
           .build())
       .addFieldViolations(BadRequest.FieldViolation.newBuilder()
           .setField("age")
           .setDescription("Must be between 1 and 150")
           .build())
       .build()))
   .build();
 
responseObserver.onError(StatusProto.toStatusRuntimeException(status));

try {
   UserResponse response = userStub.getUser(request);
} catch (StatusRuntimeException e) {
   switch (e.getStatus().getCode()) {
       case NOT_FOUND:
           log.warn("User not found: {}", e.getStatus().getDescription());
           break;
       case UNAVAILABLE:
           log.error("Service unavailable, will retry");
           // retry logic
           break;
       case DEADLINE_EXCEEDED:
           log.error("Request timed out");
           break;
       default:
           log.error("gRPC error: {} - {}", e.getStatus().getCode(), e.getStatus().getDescription());
   }
}

public class TraceIdServerInterceptor implements ServerInterceptor {
 
   private static final Metadata.Key<String> TRACE_ID_KEY =
       Metadata.Key.of("x-trace-id", Metadata.ASCII_STRING_MARSHALLER);
 
   public static final Context.Key<String> TRACE_ID_CONTEXT_KEY = Context.key("traceId");
 
   @Override
   public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
           ServerCall<ReqT, RespT> call,
           Metadata headers,
           ServerCallHandler<ReqT, RespT> next) {
 
       String traceId = headers.get(TRACE_ID_KEY);
       if (traceId == null) {
           traceId = UUID.randomUUID().toString();
       }
 
       // Put traceId in context for downstream use
       Context ctx = Context.current().withValue(TRACE_ID_CONTEXT_KEY, traceId);
       return Contexts.interceptCall(ctx, call, headers, next);
   }
}

public class AuthClientInterceptor implements ClientInterceptor {
 
   private static final Metadata.Key<String> AUTH_KEY =
       Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
 
   private final TokenProvider tokenProvider;
 
   public AuthClientInterceptor(TokenProvider tokenProvider) {
       this.tokenProvider = tokenProvider;
   }
 
   @Override
   public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
           MethodDescriptor<ReqT, RespT> method,
           CallOptions callOptions,
           Channel next) {
 
       return new ForwardingClientCall.SimpleForwardingClientCall<>(
               next.newCall(method, callOptions)) {
           @Override
           public void start(Listener<RespT> responseListener, Metadata headers) {
               headers.put(AUTH_KEY, "Bearer " + tokenProvider.getToken());
               super.start(responseListener, headers);
           }
       };
   }
}

public class LoggingServerInterceptor implements ServerInterceptor {
 
   @Override
   public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
           ServerCall<ReqT, RespT> call,
           Metadata headers,
           ServerCallHandler<ReqT, RespT> next) {
 
       long startTime = System.nanoTime();
       String methodName = call.getMethodDescriptor().getFullMethodName();
 
       log.info("gRPC request | method={}", methodName);
 
       ServerCall<ReqT, RespT> wrappedCall = new ForwardingServerCall.SimpleForwardingServerCall<>(call) {
           @Override
           public void close(Status status, Metadata trailers) {
               long duration = (System.nanoTime() - startTime) / 1_000_000;
               log.info("gRPC response | method={} | status={} | duration={}ms",
                   methodName, status.getCode(), duration);
               super.close(status, trailers);
           }
       };
 
       return next.startCall(wrappedCall, headers);
   }
}

Server server = ServerBuilder.forPort(9090)
   .intercept(new TraceIdServerInterceptor())    // add đầu tiên → chạy sau cùng
   .intercept(new AuthServerInterceptor())       // add thứ 2 → chạy thứ 2
   .intercept(new LoggingServerInterceptor())    // add sau cùng → chạy đầu tiên
   .addService(new UserServiceImpl())
   .build();
 
// Lưu ý: interceptors chạy theo thứ tự NGƯỢC (last added = first executed)
// Thực tế: Logging → Auth → TraceId → Service
// Muốn TraceId chạy trước để log có trace id thì add nó SAU CÙNG

// Client set deadline (timeout)
UserResponse response = userStub
   .withDeadlineAfter(5, TimeUnit.SECONDS)  // timeout 5s
   .getUser(request);
 
// Server check deadline
@Override
public void getUser(GetUserRequest request, StreamObserver<UserResponse> responseObserver) {
   if (Context.current().isCancelled()) {
       responseObserver.onError(Status.CANCELLED
           .withDescription("Client cancelled or deadline exceeded")
           .asRuntimeException());
       return;
   }
   // ... process
}

// Client gửi metadata
Metadata metadata = new Metadata();
metadata.put(Metadata.Key.of("x-workspace-id", Metadata.ASCII_STRING_MARSHALLER), "ws-123");
metadata.put(Metadata.Key.of("x-request-id", Metadata.ASCII_STRING_MARSHALLER), UUID.randomUUID().toString());
 
UserResponse response = userStub
   .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata))
   .getUser(request);
 
// Server đọc metadata
@Override
public void getUser(GetUserRequest request, StreamObserver<UserResponse> responseObserver) {
   String workspaceId = WORKSPACE_ID_CTX_KEY.get();  // from interceptor context
   // ...
}

<dependencies>
   <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-grpc-server</artifactId>
   </dependency>
   <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-grpc-client</artifactId>
   </dependency>
 
   <!-- Test: in-process transport, @AutoConfigureTestGrpcTransport -->
   <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-grpc-server-test</artifactId>
       <scope>test</scope>
   </dependency>
   <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-grpc-client-test</artifactId>
       <scope>test</scope>
   </dependency>
</dependencies>
 
<build>
   <plugins>
       <!-- Sinh code Java từ src/main/proto/*.proto, version đã được Spring Boot quản lý -->
       <plugin>
           <groupId>io.github.ascopes</groupId>
           <artifactId>protobuf-maven-plugin</artifactId>
       </plugin>
       <plugin>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-maven-plugin</artifactId>
       </plugin>
   </plugins>
</build>

# application.yml
spring:
  grpc:
    server:
      port: 9090
      keepalive:
        permit:
          time: 30s # cho phép client ping mỗi 30s (mặc định chỉ cho 5 phút)
          without-calls: true
    client:
      channel:
        user-service: # tên channel, dùng lại trong code
          target: 'static://localhost:9090'
          default:
            deadline: 3s # mặc định KHÔNG có deadline
            load-balancing-policy: round_robin
          keepalive:
            time: 30s
            timeout: 5s

import org.springframework.grpc.server.service.GrpcService;
 
@GrpcService // bean Spring, tự đăng ký với gRPC server
public class UserGrpcService extends UserServiceGrpc.UserServiceImplBase {
 
   private final UserService userService;
 
   public UserGrpcService(UserService userService) {
       this.userService = userService;
   }
 
   @Override
   public void getUser(GetUserRequest request, StreamObserver<UserResponse> responseObserver) {
       if (request.getUserId().isBlank()) {
           responseObserver.onError(Status.INVALID_ARGUMENT
               .withDescription("user_id must not be empty")
               .asRuntimeException());
           return;
       }
       // NotFoundException hay lỗi bất ngờ cứ để ném ra, GrpcExceptionHandler lo việc map sang Status
       User user = userService.findById(request.getUserId());
       responseObserver.onNext(toProto(user));
       responseObserver.onCompleted();
   }
 
   private UserResponse toProto(User user) {
       return UserResponse.newBuilder()
           .setId(user.id())
           .setName(user.name())
           .setEmail(user.email())
           .build();
   }
}

import org.springframework.grpc.server.exception.GrpcExceptionHandler;
 
@Configuration
public class GrpcErrorConfig {
 
   @Bean
   GrpcExceptionHandler notFoundHandler() {
       return exception -> exception instanceof NotFoundException
           ? Status.NOT_FOUND.withDescription(exception.getMessage()).asException()
           : null; // null = không xử lý, nhường handler khác
   }
}

import org.springframework.grpc.client.GrpcChannelFactory;
 
@Service
@Slf4j
public class UserGrpcClient {
 
   private final UserServiceGrpc.UserServiceBlockingStub userStub;
 
   public UserGrpcClient(GrpcChannelFactory channels) {
       // "user-service" khớp với spring.grpc.client.channel.user-service.* trong application.yml
       this.userStub = UserServiceGrpc.newBlockingStub(channels.createChannel("user-service"));
   }
 
   public UserResponse getUser(String userId) {
       try {
           // Deadline riêng cho call này, ghi đè default.deadline của channel
           return userStub
               .withDeadlineAfter(3, TimeUnit.SECONDS)
               .getUser(GetUserRequest.newBuilder()
                   .setUserId(userId)
                   .build());
       } catch (StatusRuntimeException e) {
           log.error("gRPC call failed | status={} | desc={}",
               e.getStatus().getCode(), e.getStatus().getDescription());
           throw new ServiceCallException("User service unavailable", e);
       }
   }
}

import org.springframework.grpc.client.GlobalClientInterceptor;
import org.springframework.grpc.server.GlobalServerInterceptor;
 
@Configuration
public class GrpcInterceptorConfig {
 
   @Bean
   @Order(Ordered.HIGHEST_PRECEDENCE) // bọc ngoài cùng, thấy được cả status do GrpcExceptionHandler trả ra
   @GlobalServerInterceptor
   ServerInterceptor loggingInterceptor() {
       return new LoggingServerInterceptor();
   }
 
   @Bean
   @Order(10)
   @GlobalServerInterceptor
   ServerInterceptor traceIdInterceptor() {
       return new TraceIdServerInterceptor();
   }
 
   @Bean
   @GlobalClientInterceptor
   ClientInterceptor authClientInterceptor(TokenProvider tokenProvider) {
       return new AuthClientInterceptor(tokenProvider);
   }
}

# Ví dụ cấu hình Client-side LB với Spring Boot
spring:
  grpc:
    client:
      channel:
        user-service:
          target: 'dns:///user-service-grpc:9090' # DNS trả về nhiều IP, client tự chia tải
          default:
            load-balancing-policy: round_robin

# Headless service (trả về tất cả pod IPs)
apiVersion: v1
kind: Service
metadata:
  name: user-service-grpc
spec:
  clusterIP: None # headless
  ports:
    - port: 9090
      protocol: TCP
  selector:
    app: user-service

spring:
  ssl:
    bundle:
      pem:
        grpc-server:
          keystore:
            certificate: 'file:certs/server.crt'
            private-key: 'file:certs/server.key'
        grpc-client:
          truststore:
            certificate: 'file:certs/ca.crt'
  grpc:
    # Server
    server:
      ssl:
        bundle: grpc-server
    # Client
    client:
      channel:
        user-service:
          target: 'static://user-service:9090'
          ssl:
            bundle: grpc-client

spring:
  ssl:
    bundle:
      pem:
        grpc-server:
          keystore:
            certificate: 'file:certs/server.crt'
            private-key: 'file:certs/server.key'
          truststore:
            certificate: 'file:certs/ca.crt'
        grpc-client:
          keystore:
            certificate: 'file:certs/client.crt'
            private-key: 'file:certs/client.key'
          truststore:
            certificate: 'file:certs/ca.crt'
  grpc:
    server:
      ssl:
        bundle: grpc-server
        client-auth: require
    client:
      channel:
        user-service:
          ssl:
            bundle: grpc-client

public class JwtAuthServerInterceptor implements ServerInterceptor {
 
   private static final Metadata.Key<String> AUTH_KEY =
       Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
 
   private final JwtValidator jwtValidator;
 
   public JwtAuthServerInterceptor(JwtValidator jwtValidator) {
       this.jwtValidator = jwtValidator;
   }
 
   @Override
   public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
           ServerCall<ReqT, RespT> call,
           Metadata headers,
           ServerCallHandler<ReqT, RespT> next) {
 
       String authHeader = headers.get(AUTH_KEY);
 
       if (authHeader == null || !authHeader.startsWith("Bearer ")) {
           call.close(Status.UNAUTHENTICATED.withDescription("Missing token"), new Metadata());
           return new ServerCall.Listener<>() {};
       }
 
       String token = authHeader.substring(7);
       try {
           Claims claims = jwtValidator.validate(token);
           Context ctx = Context.current()
               .withValue(USER_ID_CTX_KEY, claims.getSubject())
               .withValue(ROLES_CTX_KEY, claims.get("roles", List.class));
           return Contexts.interceptCall(ctx, call, headers, next);
       } catch (JwtException e) {
           call.close(Status.UNAUTHENTICATED.withDescription("Invalid token"), new Metadata());
           return new ServerCall.Listener<>() {};
       }
   }
}

// Reuse channel (thread-safe, multiplexed)
// KHÔNG tạo channel mới cho mỗi request
ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 9090)
   .usePlaintext()
   .keepAliveTime(30, TimeUnit.SECONDS)
   .keepAliveTimeout(5, TimeUnit.SECONDS)
   .maxInboundMessageSize(10 * 1024 * 1024)  // 10MB
   .build();
 
// Tạo stub từ channel (lightweight, có thể tạo nhiều)
UserServiceGrpc.UserServiceBlockingStub stub = UserServiceGrpc.newBlockingStub(channel);

// Server
ServerBuilder.forPort(9090)
   .maxInboundMessageSize(10 * 1024 * 1024)  // 10MB max request
   .addService(new UserServiceImpl())
   .build();
 
// Client
ManagedChannelBuilder.forAddress("localhost", 9090)
   .maxInboundMessageSize(10 * 1024 * 1024)  // 10MB max response
   .build();

// Client enable compression
UserResponse response = userStub
   .withCompression("gzip")
   .getUser(request);
 
// Server tự giải nén request nếu client gửi compressed (registry mặc định đã có gzip).
// Response thì server KHÔNG tự nén, phải bật cho từng call:
@Override
public void getUser(GetUserRequest request, StreamObserver<UserResponse> responseObserver) {
   ((ServerCallStreamObserver<UserResponse>) responseObserver).setCompression("gzip");
   // ... onNext / onCompleted như bình thường
}

// Server-side flow control cho streaming
@Override
public void listUsers(ListUsersRequest request, StreamObserver<UserResponse> responseObserver) {
   ServerCallStreamObserver<UserResponse> serverObserver =
       (ServerCallStreamObserver<UserResponse>) responseObserver;
 
   // Set callback khi client sẵn sàng nhận thêm
   serverObserver.setOnReadyHandler(() -> {
       while (serverObserver.isReady()) {
           UserResponse next = getNextUser();
           if (next == null) {
               serverObserver.onCompleted();
               return;
           }
           serverObserver.onNext(next);
       }
       // Client chưa ready → pause, sẽ được gọi lại khi ready
   });
}

# Mỗi channel tự giữ connection tới từng backend bên dưới (subchannel),
# nên dùng lại một channel cho cả ứng dụng là đủ, không cần pool riêng.
# Nhiều backend cố định thì để DNS trả về nhiều IP (headless service),
# hoặc đặt load balancer L7 phía trước.
spring:
  grpc:
    client:
      channel:
        user-service:
          target: 'dns:///user-service-grpc:9090'
          default:
            load-balancing-policy: round_robin
          keepalive:
            time: 30s
            timeout: 5s
            without-calls: true

// Version 1
message UserResponse {
 string id = 1;
 string name = 2;
}
 
// Version 2 (backward compatible)
message UserResponse {
 string id = 1;
 string name = 2;
 string email = 3;          // NEW: old clients ignore unknown fields
 // int32 age = 4;          // REMOVED: just stop using, don't reuse number
 reserved 4;                // Prevent reuse of field number 4
 reserved "age";            // Prevent reuse of field name "age"
}

// Messages: PascalCase
message UserProfile {}
 
// Fields: snake_case
string user_name = 1;
int32 page_size = 2;
 
// Enums: SCREAMING_SNAKE_CASE, prefix với type name
enum UserStatus {
 USER_STATUS_UNSPECIFIED = 0;  // always have 0 = unspecified
 USER_STATUS_ACTIVE = 1;
}
 
// Services: PascalCase
service UserService {}
 
// RPCs: PascalCase
rpc GetUser(GetUserRequest) returns (UserResponse);
 
// Files: snake_case.proto
// user_service.proto

import "google/protobuf/timestamp.proto";
import "google/protobuf/wrappers.proto";
import "google/protobuf/empty.proto";
 
message AuditableEntity {
 google.protobuf.Timestamp created_at = 1;
 google.protobuf.Timestamp updated_at = 2;
 string created_by = 3;
}
 
// Wrapper types cho nullable primitives
// (từ protobuf 3.15 có thể dùng `optional int32 max_results = 1;` thay cho wrapper)
message UserPreferences {
 google.protobuf.Int32Value max_results = 1;  // nullable int
 google.protobuf.BoolValue notifications_enabled = 2;  // nullable bool
}
 
// Empty request/response
rpc Ping(google.protobuf.Empty) returns (google.protobuf.Empty);
 
// Pagination pattern
message PageRequest {
 int32 page_size = 1;
 string page_token = 2;
}
 
message PageResponse {
 string next_page_token = 1;
 int32 total_count = 2;
}
 
message ListUsersResponse {
 repeated UserResponse users = 1;
 PageResponse page = 2;
}

@ExtendWith(MockitoExtension.class)
class UserGrpcServiceTest {
 
   @Mock
   private UserService userService;
 
   @InjectMocks
   private UserGrpcService grpcService;
 
   @Test
   void getUser_success() {
       // Given
       User user = new User("123", "John", "john@test.com");
       when(userService.findById("123")).thenReturn(user);
 
       StreamRecorder<UserResponse> recorder = StreamRecorder.create();
       GetUserRequest request = GetUserRequest.newBuilder().setUserId("123").build();
 
       // When
       grpcService.getUser(request, recorder);
 
       // Then
       assertNull(recorder.getError());
       List<UserResponse> results = recorder.getValues();
       assertEquals(1, results.size());
       assertEquals("123", results.get(0).getId());
       assertEquals("John", results.get(0).getName());
   }
 
   @Test
   void getUser_notFound() {
       when(userService.findById("999")).thenThrow(new NotFoundException("User not found"));
 
       StreamRecorder<UserResponse> recorder = StreamRecorder.create();
       GetUserRequest request = GetUserRequest.newBuilder().setUserId("999").build();
 
       grpcService.getUser(request, recorder);
 
       assertNotNull(recorder.getError());
       assertEquals(Status.NOT_FOUND.getCode(),
           Status.fromThrowable(recorder.getError()).getCode());
   }
}

import org.springframework.boot.grpc.test.autoconfigure.AutoConfigureTestGrpcTransport;
import org.springframework.grpc.client.ImportGrpcClients;
 
@SpringBootTest
@AutoConfigureTestGrpcTransport
@ImportGrpcClients(types = UserServiceGrpc.UserServiceBlockingStub.class)
class UserGrpcServiceIntegrationTest {
 
   @Autowired
   private UserServiceGrpc.UserServiceBlockingStub userStub;
 
   @Test
   void getUser_integration() {
       GetUserRequest request = GetUserRequest.newBuilder()
           .setUserId("123")
           .build();
 
       UserResponse response = userStub.getUser(request);
 
       assertNotNull(response);
       assertEquals("123", response.getId());
   }
 
   @Test
   void getUser_invalidArgument() {
       GetUserRequest request = GetUserRequest.newBuilder()
           .setUserId("")
           .build();
 
       StatusRuntimeException exception = assertThrows(
           StatusRuntimeException.class,
           () -> userStub.getUser(request)
       );
 
       assertEquals(Status.INVALID_ARGUMENT.getCode(), exception.getStatus().getCode());
   }
}

// Có actuator + micrometer-registry-prometheus là Spring Boot tự bật observation cho gRPC.
// Tên trên Prometheus (đo thật với Spring Boot 4.1.1):
// grpc_server_seconds_count / _sum / _max{rpc_service, rpc_method, rpc_type, grpc_status_code, error}
// grpc_server_received_total, grpc_server_sent_total   (số message nhận/gửi)
// grpc_client_seconds_count / _sum / _max{..., net_peer_name, net_peer_port}  (phía client)
 
// Custom metrics
@GrpcService
public class UserGrpcService extends UserServiceGrpc.UserServiceImplBase {
 
   private final Counter userLookupCounter;
   private final Timer userLookupTimer;
 
   public UserGrpcService(MeterRegistry registry) {
       this.userLookupCounter = registry.counter("grpc.user.lookup.total");
       this.userLookupTimer = registry.timer("grpc.user.lookup.duration");
   }
 
   @Override
   public void getUser(GetUserRequest request, StreamObserver<UserResponse> responseObserver) {
       userLookupCounter.increment();
       userLookupTimer.record(() -> {
           // ... implementation
       });
   }
}

// Với Spring Boot: có Micrometer Tracing (vd micrometer-tracing-bridge-otel) là trace context
// tự đi qua gRPC metadata headers, cả server lẫn client, không cần code thêm.
// Không dùng Spring thì có thể dùng opentelemetry-instrumentation-grpc.
 
// Manual span creation
@Override
public void getUser(GetUserRequest request, StreamObserver<UserResponse> responseObserver) {
   Span span = tracer.spanBuilder("UserService.getUser")
       .setAttribute("user.id", request.getUserId())
       .startSpan();
 
   try (Scope scope = span.makeCurrent()) {
       // ... implementation
       span.setStatus(StatusCode.OK);
   } catch (Exception e) {
       span.setStatus(StatusCode.ERROR, e.getMessage());
       span.recordException(e);
       throw e;
   } finally {
       span.end();
   }
}