A->B(当前)->C的场景下，将远程服务端C的RPc异常异步返回给调用者A

@see SofaResponseCallback#onSofaException(SofaRpcException, String, RequestBase)
964588
@Override
    public void sendSofaException(SofaRpcException sofaException) {
        checkState();
        SofaResponse response = new SofaResponse();
        response.setErrorMsg(sofaException.getMessage());
        sendSofaResponse(response, sofaException);
    }
294934
@Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable exception) throws Exception {
    if (exception instanceof FrameDecodingException) {
      int streamId = ((FrameDecodingException) exception).streamId;
      LOG.debug("[{}] Error while decoding response on stream id {}", logPrefix, streamId);
      if (streamId >= 0) {
        // We know which request matches the failing response, fail that one only
        ResponseCallback responseCallback = inFlight.get(streamId);
        if (responseCallback != null) {
          try {
            responseCallback.onFailure(exception.getCause());
          } catch (Throwable t) {
            Loggers.warnWithException(
                LOG, "[{}] Unexpected error while invoking failure handler", logPrefix, t);
          }
        }
        release(streamId, ctx);
      } else {
        Loggers.warnWithException(
            LOG,
            "[{}] Unexpected error while decoding incoming event frame",
            logPrefix,
            exception.getCause());
      }
    } else {
      // Otherwise fail all pending requests
      abortAllInFlight(
          (exception instanceof HeartbeatException)
              ? (HeartbeatException) exception
              : new ClosedConnectionException("Unexpected error on channel", exception));
      ctx.close();
    }
  }
490271
public void checkResponse(RpcRequest request) throws IOException {
        if (getResponse().getRejectStatus() == RejectStatus.AUTH_ERROR.getValue()) {
            request.setUsePrivilegedPort(true);
        }
        if (getResponse().getState() != NfsStatus.NFS3_OK.getValue()) {
            int responseState = getResponse().getState();
            if (responseState == NfsStatus.NFS3ERR_NOENT.getValue()) {
                throw new FileNotFoundException(request.getErrorMessage());
            }
            NfsStatus status = NfsStatus.fromValue(responseState);
            throw new NfsException(status, request.getErrorMessage() + " error code:" + status.toString());
        }
    }
1210565
@Override
    protected final void onStreamError(ChannelHandlerContext ctx, boolean outbound, Throwable cause,
                                 Http2Exception.StreamException streamException) {
        int streamId = streamException.streamId();
        Http2Stream connectionStream = connection().stream(streamId);
        if (connectionStream == null) {
            onHttp2UnknownStreamError(ctx, cause, streamException);
            // Write a RST_STREAM
            super.onStreamError(ctx, outbound, cause, streamException);
            return;
        }

        Http2FrameStream stream = connectionStream.getProperty(streamKey);
        if (stream == null) {
            LOG.warn("Stream exception thrown without stream object attached.", cause);
            // Write a RST_STREAM
            super.onStreamError(ctx, outbound, cause, streamException);
            return;
        }

        if (!outbound) {
            // We only forward non outbound errors as outbound errors will already be reflected by failing the promise.
            onHttp2FrameStreamException(ctx, new Http2FrameStreamException(stream, streamException.error(), cause));
        }
    }
374342
private void handleRpcInvocation(RpcInvocation rpcInvocation) {
		Method rpcMethod = null;

		try {
			String methodName = rpcInvocation.getMethodName();
			Class<?>[] parameterTypes = rpcInvocation.getParameterTypes();

			rpcMethod = lookupRpcMethod(methodName, parameterTypes);
		} catch (ClassNotFoundException e) {
			log.error("Could not load method arguments.", e);

			RpcConnectionException rpcException = new RpcConnectionException("Could not load method arguments.", e);
			getSender().tell(new Status.Failure(rpcException), getSelf());
		} catch (IOException e) {
			log.error("Could not deserialize rpc invocation message.", e);

			RpcConnectionException rpcException = new RpcConnectionException("Could not deserialize rpc invocation message.", e);
			getSender().tell(new Status.Failure(rpcException), getSelf());
		} catch (final NoSuchMethodException e) {
			log.error("Could not find rpc method for rpc invocation.", e);

			RpcConnectionException rpcException = new RpcConnectionException("Could not find rpc method for rpc invocation.", e);
			getSender().tell(new Status.Failure(rpcException), getSelf());
		}

		if (rpcMethod != null) {
			try {
				// this supports declaration of anonymous classes
				rpcMethod.setAccessible(true);

				if (rpcMethod.getReturnType().equals(Void.TYPE)) {
					// No return value to send back
					rpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs());
				}
				else {
					final Object result;
					try {
						result = rpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs());
					}
					catch (InvocationTargetException e) {
						log.debug("Reporting back error thrown in remote procedure {}", rpcMethod, e);

						// tell the sender about the failure
						getSender().tell(new Status.Failure(e.getTargetException()), getSelf());
						return;
					}

					final String methodName = rpcMethod.getName();

					if (result instanceof CompletableFuture) {
						final CompletableFuture<?> responseFuture = (CompletableFuture<?>) result;
						sendAsyncResponse(responseFuture, methodName);
					} else {
						sendSyncResponse(result, methodName);
					}
				}
			} catch (Throwable e) {
				log.error("Error while executing remote procedure call {}.", rpcMethod, e);
				// tell the sender about the failure
				getSender().tell(new Status.Failure(e), getSelf());
			}
		}
	}
836584
protected int statusCodeForException(@Nonnull Throwable exception) {
        if (exception instanceof PApplicationException) {
            PApplicationException e = (PApplicationException) exception;
            if (e.getId() == PApplicationExceptionType.INVALID_PROTOCOL ||
                e.getId() == PApplicationExceptionType.PROTOCOL_ERROR ||
                e.getId() == PApplicationExceptionType.BAD_SEQUENCE_ID ||
                e.getId() == PApplicationExceptionType.INVALID_MESSAGE_TYPE) {
                return HttpServletResponse.SC_BAD_REQUEST;
            }
            if (e.getId() == PApplicationExceptionType.UNKNOWN_METHOD) {
                return HttpServletResponse.SC_NOT_FOUND;
            }
        }
        return HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
    }
559186
protected ResponseFuture doInvokeAsync(SofaRequest request, RpcInternalContext rpcContext, int timeoutMillis) {
        SofaResponseCallback listener = request.getSofaResponseCallback();
        if (listener != null) {
            AbstractHttpClientHandler callback = new CallbackInvokeClientHandler(transportConfig.getConsumerConfig(),
                transportConfig.getProviderInfo(), listener, request, rpcContext,
                ClassLoaderUtils.getCurrentClassLoader());
            doSend(request, callback, timeoutMillis);
            return null;
        } else {
            HttpResponseFuture future = new HttpResponseFuture(request, timeoutMillis);
            AbstractHttpClientHandler callback = new FutureInvokeClientHandler(transportConfig.getConsumerConfig(),
                transportConfig.getProviderInfo(), future, request, rpcContext,
                ClassLoaderUtils.getCurrentClassLoader());
            doSend(request, callback, timeoutMillis);
            future.setSentTime();
            return future;
        }
    }
909612
private SofaRpcException cannotFoundService(String appName, String serviceName) {
        String errorMsg = LogCodes
            .getLog(LogCodes.ERROR_PROVIDER_SERVICE_CANNOT_FOUND, serviceName);
        LOGGER.errorWithApp(appName, errorMsg);
        return new SofaRpcException(RpcErrorType.SERVER_NOT_FOUND_INVOKER, errorMsg);
    }
415232
@Override
    public void onSuccess(String response)
    throws ActivityException, ConnectionException, AdapterException {
        try {
            // set the variable value based on the unwrapped soap content
            soapResponse = getSoapResponse(response);
            Node childElem = unwrapSoapResponse(soapResponse);
            String responseXml = DomHelper.toXml(childElem);

            String responseVarName = getAttributeValue(RESPONSE_VARIABLE);
            if (responseVarName == null)
                throw new AdapterException("Missing attribute: " + RESPONSE_VARIABLE);
            String responseVarType = getParameterType(responseVarName);

            if (!VariableTranslator.isDocumentReferenceVariable(getPackage(), responseVarType))
                throw new AdapterException("Response variable must be a DocumentReference: " + responseVarName);

            if (responseVarType.equals(StringDocument.class.getName())) {
                setParameterValueAsDocument(responseVarName, responseVarType, responseXml);
            }
            else {
                com.centurylink.mdw.variable.VariableTranslator varTrans = VariableTranslator.getTranslator(getPackage(), responseVarType);
                if (!(varTrans instanceof XmlDocumentTranslator))
                    throw new AdapterException("Unsupported response variable type: " + responseVarType + " (must implement XmlDocumentTranslator)");
                XmlDocumentTranslator xmlDocTrans = (XmlDocumentTranslator) varTrans;
                Object responseObj = xmlDocTrans.fromDomNode(childElem);
                setParameterValueAsDocument(responseVarName, responseVarType, responseObj);
            }
        }
        catch (Exception ex) {
            throw new ActivityException(ex.getMessage(), ex);
        }
    }
944185
@Override
    @FFDCIgnore({ IOException.class })
    public void run() {

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
            Tr.debug(tc, "Run WriteListenerRunnable start, WriteListener enabled: " + this._listener +" , current thread -->"+ Thread.currentThread().getName()); 
        }
        //clean up everything on this thread
        WebContainerRequestState reqState = WebContainerRequestState.getInstance(false);
        if (reqState!=null){
            reqState.init();
        }
        
        SRTServletRequestThreadData.getInstance().init(_requestDataWriteListenerThread);
        
        //Push the original thread's context onto the current thread, also save off the current thread's context
        _tcm.pushContextData();
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
            Tr.debug(tc, "Invoking the onWritePossible first time");
        }
        WebContainerRequestState.getInstance(true).setAttribute("com.ibm.ws.webcontainer.WriteAllowedonThisThread", true); 
        if(_hout != null){
            synchronized(_hout) {
                try {     
                    //call onWritePossible
                    this._listener.onWritePossible();
                    
                } catch (Exception e){
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){
                        Tr.debug(tc, "An exception occurred during the onWritePossible : " + e); 
                    }
                    _hout.setExceptionDuringOnWP(true);
                    _cb.error(_hout.getVc(), e );
                }
                finally{
                    //Revert back to the thread's current context
                    _tcm.popContextData();
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){
                        Tr.debug(tc, "Run WriteListenerRunnable done");
                    }
                }
            }
        }
        else{
            try {     
                //call onWritePossible
                this._listener.onWritePossible();

            } catch (IOException ioe){
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){
                    Tr.debug(tc, "An exception occurred during the onWritePossible : " + ioe); 
                }
                this._listener.onError(ioe);
            }
            finally{
                //Revert back to the thread's current context
                _tcm.popContextData();
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){
                    Tr.debug(tc, "Run WriteListenerRunnable done");
                }
            }
        }
    }
127014
@Override
    public void onRPCError(int handle, int code, String text) {
        if (handle == asyncHandle) {
            asyncHandle = 0;
            ZKUtil.fireEvent(new AsyncRPCErrorEvent(rpcName, target, code, text, handle));
            rpcName = null;
        }
    }
1187894
@Override
    public Object onText(WsRequest wsRequest, String text, ChannelContext channelContext) throws Exception {
        log.debug("onText");
        TioWebsocketMethodMapper onText = methods.getOnText();
        if (onText != null) {
            onText.getMethod().invoke(onText.getInstance(), channelContext, text);
        } else {
            TioWebsocketMethodMapper onBeforeText = methods.getOnBeforeText();
            if (onBeforeText != null) {
                TioWebsocketRequest invoke = (TioWebsocketRequest) onBeforeText.getMethod().invoke(onBeforeText.getInstance(), channelContext, text);
                onMapEvent(invoke, channelContext);
            }
        }
        return null;
    }
1290489
private SofaRpcException clientTimeoutWhenSendResponse(String appName, String serviceName, String remoteAddress) {
        String errorMsg = LogCodes.getLog(
            LogCodes.ERROR_DISCARD_TIMEOUT_RESPONSE, serviceName, remoteAddress);
        if (LOGGER.isWarnEnabled(appName)) {
            LOGGER.warnWithApp(appName, errorMsg);
        }
        return new SofaRpcException(RpcErrorType.SERVER_UNDECLARED_ERROR, errorMsg);
    }
1207973
protected ActionResponse asDBFluteApplicationException(ActionRuntime runtime, RuntimeException cause) {
        final ActionResponse response;
        if (cause instanceof EntityAlreadyDeletedException) {
            response = handleEntityAlreadyDeletedException(runtime, (EntityAlreadyDeletedException) cause);
        } else if (cause instanceof EntityAlreadyUpdatedException) {
            response = handleEntityAlreadyUpdatedException(runtime, (EntityAlreadyUpdatedException) cause);
        } else if (cause instanceof EntityAlreadyExistsException) {
            response = handleEntityAlreadyExistsException(runtime, (EntityAlreadyExistsException) cause);
        } else {
            response = ActionResponse.undefined();
        }
        return response;
    }
842155
@Override
    public void error(VirtualConnection vc, Throwable t) {

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())    
            Tr.debug(tc, "error callback called , WriteListener enabled: " + this._wl); 

        // reset the reqState
        WebContainerRequestState reqState = WebContainerRequestState.getInstance(true);                         
        reqState.init();  

        synchronized(_hout._writeReadyLockObj){
            // Make sure the ready is set to true before calling onError
            _hout.set_internalReady(true);            
            _hout.setWriteReady(true);   
            reqState.setAttribute("com.ibm.ws.webcontainer.AllowWriteFromE", true); 
            reqState.setAttribute("com.ibm.ws.webcontainer.WriteAllowedonThisThread", true);
        }
        
        SRTServletRequestThreadData.getInstance().init(_requestDataAsyncWriteCallbackThread);
        
        //Push the original thread's context onto the current thread, also save off the current thread's context
        tcm.pushContextData();

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
            Tr.debug(tc, "WriteListener enabled: " + this._wl +" , calling user's onError : " + vc );            
        }       
        try{  
            synchronized(_hout) {
                //An error occurred. Issue the onError call on the user's WriteListener
                _wl.onError(t);
            }
            
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()){  
                Tr.debug(tc, "WriteListener enabled: " + this._wl +" , returned from user's onError : " + vc );            
            }  
        } catch (Exception e) {
             
            Tr.error(tc, "writeListener.onError.failed", new Object[] {this._wl, e.toString()});
            
        }
        finally{

            tcm.popContextData();//Revert back to the thread's current context

            reqState.removeAttribute("com.ibm.ws.webcontainer.AllowWriteFromE");
        }
    }
510310
protected ResponseFuture doInvokeAsync(SofaRequest request, RpcInternalContext rpcContext,
                                           InvokeContext invokeContext, int timeoutMillis)
        throws RemotingException, InterruptedException {
        SofaResponseCallback listener = request.getSofaResponseCallback();
        if (listener != null) {
            // callback调用
            InvokeCallback callback = new BoltInvokerCallback(transportConfig.getConsumerConfig(),
                transportConfig.getProviderInfo(), listener, request, rpcContext,
                ClassLoaderUtils.getCurrentClassLoader());
            // 发起调用
            RPC_CLIENT.invokeWithCallback(url, request, invokeContext, callback, timeoutMillis);
            return null;
        } else {
            // future 转为 callback
            BoltResponseFuture future = new BoltResponseFuture(request, timeoutMillis);
            InvokeCallback callback = new BoltFutureInvokeCallback(transportConfig.getConsumerConfig(),
                transportConfig.getProviderInfo(), future, request, rpcContext,
                ClassLoaderUtils.getCurrentClassLoader());
            // 发起调用
            RPC_CLIENT.invokeWithCallback(url, request, invokeContext, callback, timeoutMillis);
            future.setSentTime();
            return future;
        }
    }
737189
@Override
    public void onReceive(Object message) throws Exception {
        try {
            if (message instanceof RequestWorkerMsgType) {
                switch ((RequestWorkerMsgType) message) {
                case PROCESS_REQUEST:
                    tryCount++;

                    if (tryCount == 1) {
                        sender = getSender();

                        ConnectionlessBootstrap udpClient = bootStrapUdpClient();

                        // Start the connection attempt.
                        ChannelFuture future = udpClient
                                .connect(new InetSocketAddress(targetHost,
                                        udpMeta.getUdpPort()));

                        // first schedule timeout before the sync wait
                        timeoutDuration = Duration.create(
                                actorMaxOperationTimeoutSec, TimeUnit.SECONDS);

                        timeoutMessageCancellable = getContext()
                                .system()
                                .scheduler()
                                .scheduleOnce(
                                        timeoutDuration,
                                        getSelf(),
                                        RequestWorkerMsgType.PROCESS_ON_TIMEOUT,
                                        getContext().system().dispatcher(),
                                        getSelf());

                        // Wait until the connection attempt succeeds or fails.
                        channel = future.awaitUninterruptibly().getChannel();
                        ChannelFuture requestFuture = null;

                        // Sends the line to server need. line ending
                        requestFuture = channel.write(udpMeta.getCommand()
                                + "\r\n");

                        // Wait until all messages are flushed before closing
                        // the channel.
                        if (requestFuture != null) {
                            requestFuture.await();
                        }
                    } else {
                        getLogger().error(
                                "duplicated PROCESS_REQUEST msg. ignore...");
                    }
                    break;

                case CANCEL:
                    if (sender == null)
                        sender = getSender();
                    getLogger().info(
                            "UDP Request was CANCELLED.................{}",
                            targetHost);
                    cancelCancellable();
                    reply(null, true, PcConstants.REQUEST_CANCELED,
                            PcConstants.REQUEST_CANCELED, PcConstants.NA,
                            PcConstants.NA_INT);
                    break;

                case PROCESS_ON_EXCEPTION:

                    String errorSummary = PcErrorMsgUtils.replaceErrorMsg(cause
                            .toString());
                    String stackTrace = PcStringUtils.printStackTrace(cause);
                    cancelCancellable();
                    reply(null, true, errorSummary, stackTrace, PcConstants.NA,
                            PcConstants.NA_INT);

                    break;

                case PROCESS_ON_TIMEOUT:

                    getLogger().error("PROCESS_ON_TIMEOUT.................{}",
                            targetHost);
                    cancelCancellable();

                    String errorMsg = String
                            .format("UdpWorker Timedout after %d SEC (no response but no exception catched). Check URL: may be very slow or stuck.",
                                    actorMaxOperationTimeoutSec);

                    reply(null, true, errorMsg, errorMsg, PcConstants.NA,
                            PcConstants.NA_INT);
                    break;

                case CHECK_FUTURE_STATE:
                default:
                    sender = getSender();
                    this.cause = new ActorMessageTypeInvalidException(
                            "ActorMessageTypeInvalidException error for on "
                                    + this.targetHost);
                    getSelf().tell(RequestWorkerMsgType.PROCESS_ON_EXCEPTION,
                            getSelf());
                    break;
                }
            } else {
                unhandled(message);
                sender = getSender();
                this.cause = new ActorMessageTypeInvalidException(
                        "ActorMessageTypeInvalidException error for UDP on "
                                + this.targetHost);
                getSelf().tell(RequestWorkerMsgType.PROCESS_ON_EXCEPTION,
                        getSelf());
            }
        } catch (Exception e) {
            this.cause = e;
            getSelf()
                    .tell(RequestWorkerMsgType.PROCESS_ON_EXCEPTION, getSelf());
        }
    }