@Test
public void testRetryableExceptionWithDelay() {
  RetryInfo retryInfo =
      RetryInfo.newBuilder()
          .setRetryDelay(Duration.newBuilder().setSeconds(22L))
          .build();

  Metadata errorMetadata = new Metadata();
  errorMetadata.put(ProtoUtils.keyForProto(RetryInfo.getDefaultInstance()), retryInfo);

  StatusRuntimeException retryableException =
      new StatusRuntimeException(Status.RESOURCE_EXHAUSTED, errorMetadata);

  assertThat(createR2dbcException(retryableException))
      .isInstanceOf(R2dbcTransientResourceException.class);
}
 

private List<MethodDescriptor<Request, Response>> getMethodDescriptor( ProviderConfig providerConfig) {
    List<MethodDescriptor<Request, Response>> result = new ArrayList<>();

    Set<String> methodNames = SofaProtoUtils.getMethodNames(providerConfig.getInterfaceId());
    for (String name : methodNames) {
        MethodDescriptor<Request, Response> methodDescriptor = MethodDescriptor.<Request, Response>newBuilder()
                .setType(MethodDescriptor.MethodType.UNARY)
                .setFullMethodName(generateFullMethodName(providerConfig.getInterfaceId(), name))
                .setSampledToLocalTracing(true)
                .setRequestMarshaller(ProtoUtils.marshaller(
                        Request.getDefaultInstance()))
                .setResponseMarshaller(ProtoUtils.marshaller(
                        Response.getDefaultInstance()))
                .build();
        result.add(methodDescriptor);
    }

    return result;
}
 

private MethodDescriptor<Message, Message> getCallMethod(final Object request) {
    final String interest = request.getClass().getName();
    final Message reqIns = Requires.requireNonNull(this.parserClasses.get(interest), "null default instance: "
                                                                                     + interest);
    return MethodDescriptor //
        .<Message, Message> newBuilder() //
        .setType(MethodDescriptor.MethodType.UNARY) //
        .setFullMethodName(MethodDescriptor.generateFullMethodName(interest, GrpcRaftRpcFactory.FIXED_METHOD_NAME)) //
        .setRequestMarshaller(ProtoUtils.marshaller(reqIns)) //
        .setResponseMarshaller(
            ProtoUtils.marshaller(this.marshallerRegistry.findResponseInstanceByRequest(interest))) //
        .build();
}
 

@SuppressWarnings("unchecked")
@Override
public void registerProcessor(final RpcProcessor processor) {
    final String interest = processor.interest();
    final Message reqIns = Requires.requireNonNull(this.parserClasses.get(interest), "null default instance: " + interest);
    final MethodDescriptor<Message, Message> method = MethodDescriptor //
            .<Message, Message>newBuilder() //
            .setType(MethodDescriptor.MethodType.UNARY) //
            .setFullMethodName(
                MethodDescriptor.generateFullMethodName(processor.interest(), GrpcRaftRpcFactory.FIXED_METHOD_NAME)) //
            .setRequestMarshaller(ProtoUtils.marshaller(reqIns)) //
            .setResponseMarshaller(ProtoUtils.marshaller(this.marshallerRegistry.findResponseInstanceByRequest(interest))) //
            .build();

    final ServerCallHandler<Message, Message> handler = ServerCalls.asyncUnaryCall(
            (request, responseObserver) -> {
                final SocketAddress remoteAddress = RemoteAddressInterceptor.getRemoteAddress();
                final Connection conn = ConnectionInterceptor.getCurrentConnection(this.closedEventListeners);

                final RpcContext rpcCtx = new RpcContext() {

                    @Override
                    public void sendResponse(final Object responseObj) {
                        try {
                            responseObserver.onNext((Message) responseObj);
                            responseObserver.onCompleted();
                        } catch (final Throwable t) {
                            LOG.warn("[GRPC] failed to send response: {}.", t);
                        }
                    }

                    @Override
                    public Connection getConnection() {
                        if (conn == null) {
                            throw new IllegalStateException("fail to get connection");
                        }
                        return conn;
                    }

                    @Override
                    public String getRemoteAddress() {
                        // Rely on GRPC's capabilities, not magic (netty channel)
                        return remoteAddress != null ? remoteAddress.toString() : null;
                    }
                };

                final RpcProcessor.ExecutorSelector selector = processor.executorSelector();
                Executor executor;
                if (selector != null && request instanceof RpcRequests.AppendEntriesRequest) {
                    final RpcRequests.AppendEntriesRequest req = (RpcRequests.AppendEntriesRequest) request;
                    final RpcRequests.AppendEntriesRequestHeader.Builder header = RpcRequests.AppendEntriesRequestHeader //
                            .newBuilder() //
                            .setGroupId(req.getGroupId()) //
                            .setPeerId(req.getPeerId()) //
                            .setServerId(req.getServerId());
                    executor = selector.select(interest, header.build());
                } else {
                    executor = processor.executor();
                }

                if (executor == null) {
                    executor = this.defaultExecutor;
                }

                if (executor != null) {
                    executor.execute(() -> processor.handleRequest(rpcCtx, request));
                } else {
                    processor.handleRequest(rpcCtx, request);
                }
            });

    final ServerServiceDefinition serviceDef = ServerServiceDefinition //
            .builder(interest) //
            .addMethod(method, handler) //
            .build();
    
    this.handlerRegistry
        .addService(ServerInterceptors.intercept(serviceDef, this.serverInterceptors.toArray(new ServerInterceptor[0])));
}