From 3f8a16f0ac04b9f6bca18d48d13f8d0d0dc1fdf4 Mon Sep 17 00:00:00 2001 From: wxbty <38374721+wxbty@users.noreply.github.com> Date: Thu, 23 Mar 2023 19:50:11 +0800 Subject: [PATCH] support exception process when service not found (#11088) --- .../common/constants/CommonConstants.java | 5 + dubbo-distribution/dubbo-all/pom.xml | 6 + .../rpc/protocol/dubbo/ByteAccessor.java | 41 ++++ .../dubbo/DecodeableRpcInvocation.java | 183 +++++++++--------- .../dubbo/rpc/protocol/dubbo/DubboCodec.java | 22 ++- 5 files changed, 164 insertions(+), 93 deletions(-) create mode 100644 dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java index 62ff4beeeb..6a723f78a4 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java @@ -620,5 +620,10 @@ public interface CommonConstants { String ENCODE_IN_IO_THREAD_KEY = "encode.in.io"; boolean DEFAULT_ENCODE_IN_IO_THREAD = false; + /** + * @since 3.2.0 + */ + String BYTE_ACCESSOR_KEY = "byte.accessor"; + String PAYLOAD = "payload"; } diff --git a/dubbo-distribution/dubbo-all/pom.xml b/dubbo-distribution/dubbo-all/pom.xml index 1e0501e7d4..a5e7f381ea 100644 --- a/dubbo-distribution/dubbo-all/pom.xml +++ b/dubbo-distribution/dubbo-all/pom.xml @@ -1004,6 +1004,12 @@ META-INF/dubbo/internal/org.apache.dubbo.remoting.Transporter + + + META-INF/dubbo/internal/org.apache.dubbo.rpc.protocol.dubbo.ByteAccessor + + diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java new file mode 100644 index 0000000000..e0913486da --- /dev/null +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java @@ -0,0 +1,41 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dubbo.rpc.protocol.dubbo; + +import org.apache.dubbo.common.extension.SPI; +import org.apache.dubbo.remoting.Channel; +import org.apache.dubbo.remoting.exchange.Request; + +import java.io.InputStream; + +import static org.apache.dubbo.common.extension.ExtensionScope.FRAMEWORK; + +/** + * Extension of Byte Accessor, holding attributes as RpcInvocation objects + * so that the decoded service can be used more flexibly + * @since 3.2.0 + */ +@SPI(scope = FRAMEWORK) +public interface ByteAccessor { + + /** + * Get an enhanced DecodeableRpcInvocation subclass to allow custom decode. + * The parameters are the same as {@link DecodeableRpcInvocation} + */ + DecodeableRpcInvocation getRpcInvocation(Channel channel, Request req, InputStream is, byte proto); +} diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java index c133913f9b..76b3f5ce76 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java @@ -16,7 +16,6 @@ */ package org.apache.dubbo.rpc.protocol.dubbo; - import org.apache.dubbo.common.utils.CacheableSupplier; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; @@ -60,21 +59,21 @@ import static org.apache.dubbo.rpc.Constants.SERIALIZATION_SECURITY_CHECK_KEY; public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Decodeable { - private static final ErrorTypeAwareLogger log = LoggerFactory.getErrorTypeAwareLogger(DecodeableRpcInvocation.class); + protected static final ErrorTypeAwareLogger log = LoggerFactory.getErrorTypeAwareLogger(DecodeableRpcInvocation.class); - private final Channel channel; + protected final transient Channel channel; - private final byte serializationType; + protected final byte serializationType; - private final InputStream inputStream; + protected final transient InputStream inputStream; - private final Request request; + protected final transient Request request; - private volatile boolean hasDecoded; + protected volatile boolean hasDecoded; protected final FrameworkModel frameworkModel; - private final Supplier callbackServiceCodecFactory; + protected final transient Supplier callbackServiceCodecFactory; public DecodeableRpcInvocation(FrameworkModel frameworkModel, Channel channel, Request request, InputStream is, byte id) { this.frameworkModel = frameworkModel; @@ -85,7 +84,7 @@ public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Dec this.request = request; this.inputStream = is; this.serializationType = id; - this.callbackServiceCodecFactory = CacheableSupplier.newSupplier(()-> + this.callbackServiceCodecFactory = CacheableSupplier.newSupplier(() -> new CallbackServiceCodec(frameworkModel)); } @@ -111,10 +110,6 @@ public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Dec throw new UnsupportedOperationException(); } - private void checkSerializationTypeFromRemote() { - - } - @Override public Object decode(Channel channel, InputStream input) throws IOException { ObjectInput in = CodecSupport.getSerialization(serializationType) @@ -143,78 +138,14 @@ public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Dec Object[] args = DubboCodec.EMPTY_OBJECT_ARRAY; Class[] pts = DubboCodec.EMPTY_CLASS_ARRAY; if (desc.length() > 0) { -// if (RpcUtils.isGenericCall(path, getMethodName()) || RpcUtils.isEcho(path, getMethodName())) { -// pts = ReflectUtils.desc2classArray(desc); -// } else { - FrameworkServiceRepository repository = frameworkModel.getServiceRepository(); - List providerModels = repository.lookupExportedServicesWithoutGroup(keyWithoutGroup(path, version)); - ServiceDescriptor serviceDescriptor = null; - if (CollectionUtils.isNotEmpty(providerModels)) { - for (ProviderModel providerModel : providerModels) { - serviceDescriptor = providerModel.getServiceModel(); - if (serviceDescriptor != null) { - break; - } - } - } - if (serviceDescriptor == null) { - // Unable to find ProviderModel from Exported Services - for (ApplicationModel applicationModel : frameworkModel.getApplicationModels()) { - for (ModuleModel moduleModel : applicationModel.getModuleModels()) { - serviceDescriptor = moduleModel.getServiceRepository().lookupService(path); - if (serviceDescriptor != null) { - break; - } - } - } - } - - if (serviceDescriptor != null) { - MethodDescriptor methodDescriptor = serviceDescriptor.getMethod(getMethodName(), desc); - if (methodDescriptor != null) { - pts = methodDescriptor.getParameterClasses(); - this.setReturnTypes(methodDescriptor.getReturnTypes()); - - // switch TCCL - if (CollectionUtils.isNotEmpty(providerModels)) { - if (providerModels.size() == 1) { - Thread.currentThread().setContextClassLoader(providerModels.get(0).getClassLoader()); - } else { - // try all providerModels' classLoader can load pts, use the first one - for (ProviderModel providerModel : providerModels) { - ClassLoader classLoader = providerModel.getClassLoader(); - boolean match = true; - for (Class pt : pts) { - try { - if (!pt.equals(classLoader.loadClass(pt.getName()))) { - match = false; - } - } catch (ClassNotFoundException e) { - match = false; - } - } - if (match) { - Thread.currentThread().setContextClassLoader(classLoader); - break; - } - } - } - } - } - } - + pts = drawPts(path, version, desc, pts); if (pts == DubboCodec.EMPTY_CLASS_ARRAY) { if (!RpcUtils.isGenericCall(desc, getMethodName()) && !RpcUtils.isEcho(desc, getMethodName())) { throw new IllegalArgumentException("Service not found:" + path + ", " + getMethodName()); } pts = ReflectUtils.desc2classArray(desc); } -// } - - args = new Object[pts.length]; - for (int i = 0; i < args.length; i++) { - args[i] = in.readObject(pts[i]); - } + args = drawArgs(in, pts); } setParameterTypes(pts); @@ -223,17 +154,7 @@ public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Dec addObjectAttachments(map); } - //decode argument ,may be callback - CallbackServiceCodec callbackServiceCodec = callbackServiceCodecFactory.get(); - for (int i = 0; i < args.length; i++) { - args[i] = callbackServiceCodec.decodeInvocationArgument(channel, this, pts, i, args[i]); - } - - setArguments(args); - String targetServiceName = buildKey(getAttachment(PATH_KEY), - getAttachment(GROUP_KEY), - getAttachment(VERSION_KEY)); - setTargetServiceUniqueName(targetServiceName); + decodeArgument(channel, pts, args); } catch (ClassNotFoundException e) { throw new IOException(StringUtils.toString("Read invocation data failed.", e)); } finally { @@ -245,4 +166,86 @@ public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Dec return this; } + + protected void decodeArgument(Channel channel, Class[] pts, Object[] args) throws IOException { + CallbackServiceCodec callbackServiceCodec = callbackServiceCodecFactory.get(); + for (int i = 0; i < args.length; i++) { + args[i] = callbackServiceCodec.decodeInvocationArgument(channel, this, pts, i, args[i]); + } + + setArguments(args); + String targetServiceName = buildKey(getAttachment(PATH_KEY), + getAttachment(GROUP_KEY), + getAttachment(VERSION_KEY)); + setTargetServiceUniqueName(targetServiceName); + } + + protected Class[] drawPts(String path, String version, String desc, Class[] pts) { + FrameworkServiceRepository repository = frameworkModel.getServiceRepository(); + List providerModels = repository.lookupExportedServicesWithoutGroup(keyWithoutGroup(path, version)); + ServiceDescriptor serviceDescriptor = null; + if (CollectionUtils.isNotEmpty(providerModels)) { + for (ProviderModel providerModel : providerModels) { + serviceDescriptor = providerModel.getServiceModel(); + if (serviceDescriptor != null) { + break; + } + } + } + if (serviceDescriptor == null) { + // Unable to find ProviderModel from Exported Services + for (ApplicationModel applicationModel : frameworkModel.getApplicationModels()) { + for (ModuleModel moduleModel : applicationModel.getModuleModels()) { + serviceDescriptor = moduleModel.getServiceRepository().lookupService(path); + if (serviceDescriptor != null) { + break; + } + } + } + } + + if (serviceDescriptor != null) { + MethodDescriptor methodDescriptor = serviceDescriptor.getMethod(getMethodName(), desc); + if (methodDescriptor != null) { + pts = methodDescriptor.getParameterClasses(); + this.setReturnTypes(methodDescriptor.getReturnTypes()); + + // switch TCCL + if (CollectionUtils.isNotEmpty(providerModels)) { + if (providerModels.size() == 1) { + Thread.currentThread().setContextClassLoader(providerModels.get(0).getClassLoader()); + } else { + // try all providerModels' classLoader can load pts, use the first one + for (ProviderModel providerModel : providerModels) { + ClassLoader classLoader = providerModel.getClassLoader(); + boolean match = true; + for (Class pt : pts) { + try { + if (!pt.equals(classLoader.loadClass(pt.getName()))) { + match = false; + } + } catch (ClassNotFoundException e) { + match = false; + } + } + if (match) { + Thread.currentThread().setContextClassLoader(classLoader); + break; + } + } + } + } + } + } + return pts; + } + + protected Object[] drawArgs(ObjectInput in, Class[] pts) throws IOException, ClassNotFoundException { + Object[] args; + args = new Object[pts.length]; + for (int i = 0; i < args.length; i++) { + args[i] = in.readObject(pts[i]); + } + return args; + } } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java index 88a0f5622d..0a7f96745c 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java @@ -42,9 +42,11 @@ import org.apache.dubbo.rpc.model.FrameworkModel; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_VERSION_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.BYTE_ACCESSOR_KEY; import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_ISOLATION; import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY; import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY; @@ -73,10 +75,15 @@ public class DubboCodec extends ExchangeCodec { private static final AtomicBoolean decodeInUserThreadLogged = new AtomicBoolean(false); private final CallbackServiceCodec callbackServiceCodec; private final FrameworkModel frameworkModel; + private final ByteAccessor customByteAccessor; public DubboCodec(FrameworkModel frameworkModel) { this.frameworkModel = frameworkModel; callbackServiceCodec = new CallbackServiceCodec(frameworkModel); + customByteAccessor = Optional.ofNullable(System.getProperty(BYTE_ACCESSOR_KEY)) + .filter(StringUtils::isNotBlank) + .map(key -> frameworkModel.getExtensionLoader(ByteAccessor.class).getExtension(key)) + .orElse(null); } @Override @@ -153,11 +160,20 @@ public class DubboCodec extends ExchangeCodec { req = new HeartBeatRequest(id); DecodeableRpcInvocation inv; if (isDecodeDataInIoThread(channel)) { - inv = new DecodeableRpcInvocation(frameworkModel, channel, req, is, proto); + if (customByteAccessor != null) { + inv = customByteAccessor.getRpcInvocation(channel, req, is, proto); + } else { + inv = new DecodeableRpcInvocation(frameworkModel, channel, req, is, proto); + } inv.decode(); } else { - inv = new DecodeableRpcInvocation(frameworkModel, channel, req, - new UnsafeByteArrayInputStream(readMessageData(is)), proto); + if (customByteAccessor != null) { + inv = customByteAccessor.getRpcInvocation(channel, req, + new UnsafeByteArrayInputStream(readMessageData(is)), proto); + } else { + inv = new DecodeableRpcInvocation(frameworkModel, channel, req, + new UnsafeByteArrayInputStream(readMessageData(is)), proto); + } } data = inv; }