grpc-java双向流RPC：服务端无法接收客户端自定义异常

问题场景 #

在双向流RPC场景下，客户端调用outbound StreamObserver.onError时，传入构造好的INVALID_ARGUMENT类型StatusRuntimeException，但服务端通过inbound StreamObserver.onError接收到的却是io.grpc.StatusRuntimeException: CANCELLED: client canceled，而非预期的异常，且元数据为空。

客户端代码 #

@Override
public void onNext(@NonNull ActionReq req) {
    String actionId = req.getId();
    String actionName = req.getName();
    Log.d(TAG, String.format("onNext: id:%s,name:%s", actionId, actionName));
    if (!mActionId.equals(actionId)) {
        Metadata metadata = new Metadata();
        metadata.put(GrpcKeyUtil.sMetaActorActionIdKey, mActionId);
        StatusRuntimeException e = Status.INVALID_ARGUMENT
                .withDescription("actionId is error! stream.id:" + mActionId + ",req.id:" + actionId)
                .asRuntimeException(metadata);
        // mPeer是客户端的outbound StreamObserver
        mPeer.onError(e);
        return;
    }
    // ...
}

服务端代码 #

@Override
public void onError(Throwable t) {
    Log.e(TAG, "onError: ", t);
    if (t instanceof StatusRuntimeException e) {
        Metadata metadata = e.getTrailers();
        Log.d(TAG, "onError: metadata " + metadata);
    }
    actionResult.onError(t);
}

原因分析 #

gRPC客户端在**接收服务端消息的onNext回调中直接调用outbound的onError**时，会触发客户端默认的流取消逻辑：此时客户端的inbound流仍处于活跃状态，gRPC会将该错误触发行为判定为主动取消整个RPC连接，因此向服务端发送CANCELLED状态，覆盖自定义的INVALID_ARGUMENT异常。

另外，如果调用onError后仍执行其他流操作（如onNext/onCompleted），也会导致自定义异常被覆盖，服务端仅收到取消状态。

解决方案 #

1. 调整错误触发时机，避免在inbound回调中直接调用outbound.onError #

不要在客户端接收服务端消息的onNext回调内直接触发outbound的错误，可将错误处理逻辑放到独立线程或异步回调中，确保客户端流状态稳定后再发送自定义异常：

@Override
public void onNext(@NonNull ActionReq req) {
    String actionId = req.getId();
    String actionName = req.getName();
    Log.d(TAG, String.format("onNext: id:%s,name:%s", actionId, actionName));
    if (!mActionId.equals(actionId)) {
        // 切换到独立线程处理错误，避免在inbound回调中触发流取消逻辑
        Executors.newSingleThreadExecutor().execute(() -> {
            Metadata metadata = new Metadata();
            metadata.put(GrpcKeyUtil.sMetaActorActionIdKey, mActionId);
            StatusRuntimeException e = Status.INVALID_ARGUMENT
                    .withDescription("actionId is error! stream.id:" + mActionId + ",req.id:" + actionId)
                    .asRuntimeException(metadata);
            mPeer.onError(e);
        });
        return;
    }
    // ...
}

2. 确保错误调用是outbound StreamObserver的最终操作 #

调用onError后，绝对不能再对同一个StreamObserver调用onNext或onCompleted，否则会触发gRPC的未定义行为，导致自定义异常被覆盖。

3. 验证元数据Key的一致性 #

确保客户端和服务端使用的Metadata.Key完全一致（包括名称、类型和编码方式），否则服务端无法解析到传递的元数据：

// 客户端和服务端需共用相同的Key定义
public static final Metadata.Key<String> sMetaActorActionIdKey =
        Metadata.Key.of("actor-action-id", Metadata.ASCII_STRING_MARSHALLER);

4. 禁用客户端不必要的自动取消配置 #

检查客户端Stub的配置，若开启了withWaitForReady()等可能触发自动取消的选项，需调整配置，确保自定义异常能正常传递。

内容的提问来源于stack exchange，提问作者Seven Sir