fix triple stream deserialize issue when in serverless scene

Author: liujianjun.ljj

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/message/triple/stream/ClientStreamObserverAdapter.java

@@ -35,9 +35,13 @@ public class ClientStreamObserverAdapter implements StreamObserver<triple.Respon
 
     private volatile Class<?>                returnType;
 
-    public ClientStreamObserverAdapter(SofaStreamObserver<Object> sofaStreamObserver, byte serializeType) {
+    private final ClassLoader                classLoader;
+
+    public ClientStreamObserverAdapter(SofaStreamObserver<Object> sofaStreamObserver, byte serializeType,
+                                       ClassLoader classLoader) {
         this.sofaStreamObserver = sofaStreamObserver;
         this.serializer = SerializerFactory.getSerializer(serializeType);
+        this.classLoader = classLoader;
     }
 
     @Override
@@ -46,14 +50,19 @@ public void onNext(triple.Response response) {
         Object appResponse = null;
         String returnTypeName = response.getType();
         if (responseData != null && responseData.length > 0) {
-            if (returnType == null && !returnTypeName.isEmpty()) {
-                try {
-                    returnType = Class.forName(returnTypeName);
-                } catch (ClassNotFoundException e) {
-                    throw new SofaRpcException(RpcErrorType.CLIENT_SERIALIZE, "Can not find return type :" + returnType);
+            ClassLoader oldClassloader = Thread.currentThread().getContextClassLoader();
+            try {
+                Thread.currentThread().setContextClassLoader(classLoader);
+                if (returnType == null && !returnTypeName.isEmpty()) {
+                    returnType = Class.forName(returnTypeName, true, classLoader);
                 }
+                appResponse = serializer.decode(new ByteArrayWrapperByteBuf(responseData), returnType, null);
+            } catch (ClassNotFoundException e) {
+                throw new SofaRpcException(RpcErrorType.CLIENT_DESERIALIZE, "Can not find return type :" + returnType,
+                    e);
+            } finally {
+                Thread.currentThread().setContextClassLoader(oldClassloader);
             }
-            appResponse = serializer.decode(new ByteArrayWrapperByteBuf(responseData), returnType, null);
         }
 
         sofaStreamObserver.onNext(appResponse);

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/transport/triple/TripleClientInvoker.java

@@ -193,7 +193,8 @@ private SofaResponse genericBinaryStreamCall(SofaRequest sofaRequest, int timeou
                 call,
                 new ClientStreamObserverAdapter(
                         sofaStreamObserver,
-                        sofaRequest.getSerializeType()
+                        sofaRequest.getSerializeType(),
+                        ClassLoaderUtils.getCurrentClassLoader()
                 )
         );
         SofaStreamObserver<Request> handler = new SofaStreamObserver() {
@@ -231,7 +232,7 @@ private SofaResponse genericServerStreamCall(SofaRequest sofaRequest, int timeou
 
         Request req = SofaProtoUtils.buildRequest(sofaRequest.getMethodArgSigs(), sofaRequest.getMethodArgs(), serialization, serializer, 1);
 
-        ClientStreamObserverAdapter responseObserver = new ClientStreamObserverAdapter(sofaStreamObserver, sofaRequest.getSerializeType());
+        ClientStreamObserverAdapter responseObserver = new ClientStreamObserverAdapter(sofaStreamObserver, sofaRequest.getSerializeType(), ClassLoaderUtils.getCurrentClassLoader());
 
         ClientCalls.asyncServerStreamingCall(call, req, responseObserver);