natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

support exception process when service not found (#11088)

This commit is contained in:
wxbty 2023-03-23 19:50:11 +08:00 committed by GitHub
parent 08f3c6b018
commit 3f8a16f0ac
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 164 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
View File

@ -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";
}

6
dubbo-distribution/dubbo-all/pom.xml
View File

@ -1004,6 +1004,12 @@
META-INF/dubbo/internal/org.apache.dubbo.remoting.Transporter
</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>
META-INF/dubbo/internal/org.apache.dubbo.rpc.protocol.dubbo.ByteAccessor
</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>

41
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ByteAccessor.java Normal file
View File

@ -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);
}

183
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DecodeableRpcInvocation.java
View File

@ -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<CallbackServiceCodec> callbackServiceCodecFactory;
protected final transient Supplier<CallbackServiceCodec> 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<ProviderModel> 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<ProviderModel> 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;
}
}

22
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java
View File

@ -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;
}