Fix serialization and deserialization classloader issues in serverless scene

Author: liujianjun.ljj

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

@@ -54,7 +54,7 @@ public ClientStreamObserverAdapter(SofaStreamObserver<Object> sofaStreamObserver
     @Override
     public void onNext(triple.Response response) {
         byte[] responseData = response.getData().toByteArray();
-        Object appResponse = null;
+        Object appResponse;
         String returnTypeName = response.getType();
         if (responseData != null && responseData.length > 0) {
             ClassLoader oldClassloader = Thread.currentThread().getContextClassLoader();
@@ -64,24 +64,35 @@ public void onNext(triple.Response response) {
                     returnType = Class.forName(returnTypeName, true, classLoader);
                 }
                 appResponse = serializer.decode(new ByteArrayWrapperByteBuf(responseData), returnType, null);
+                sofaStreamObserver.onNext(appResponse);
             } catch (ClassNotFoundException e) {
                 throw new SofaRpcException(RpcErrorType.CLIENT_DESERIALIZE, "Can not find return type :" + returnType,
                     e);
             } finally {
                 Thread.currentThread().setContextClassLoader(oldClassloader);
             }
         }
-
-        sofaStreamObserver.onNext(appResponse);
     }
 
     @Override
     public void onError(Throwable t) {
-        sofaStreamObserver.onError(t);
+        ClassLoader oldClassloader = Thread.currentThread().getContextClassLoader();
+        try {
+            Thread.currentThread().setContextClassLoader(classLoader);
+            sofaStreamObserver.onError(t);
+        } finally {
+            Thread.currentThread().setContextClassLoader(oldClassloader);
+        }
     }
 
     @Override
     public void onCompleted() {
-        sofaStreamObserver.onCompleted();
+        ClassLoader oldClassloader = Thread.currentThread().getContextClassLoader();
+        try {
+            Thread.currentThread().setContextClassLoader(classLoader);
+            sofaStreamObserver.onCompleted();
+        } finally {
+            Thread.currentThread().setContextClassLoader(oldClassloader);
+        }
     }
 }

remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/server/triple/GenericServiceImpl.java

@@ -131,20 +131,37 @@ public StreamObserver<Request> genericBiStream(StreamObserver<Response> response
 
                 @Override
                 public void onNext(Request request) {
-                    checkInitialize(request);
-                    Object message = getInvokeArgs(request, argTypes, serializer, false)[0];
-                    serverResponseHandler.setSerializeType(serializeType);
-                    clientHandler.onNext(message);
+                    try {
+                        Thread.currentThread().setContextClassLoader(serviceClassLoader);
+                        checkInitialize(request);
+                        Object message = getInvokeArgs(request, argTypes, serializer, false)[0];
+                        serverResponseHandler.setSerializeType(serializeType);
+                        clientHandler.onNext(message);
+                    } finally {
+                        Thread.currentThread().setContextClassLoader(oldClassLoader);
+                    }
                 }
 
                 @Override
                 public void onError(Throwable t) {
-                    clientHandler.onError(t);
+                    try {
+                        Thread.currentThread().setContextClassLoader(serviceClassLoader);
+                        clientHandler.onError(t);
+                    } finally {
+                        Thread.currentThread().setContextClassLoader(oldClassLoader);
+                    }
+
                 }
 
                 @Override
                 public void onCompleted() {
-                    clientHandler.onCompleted();
+                    try {
+                        Thread.currentThread().setContextClassLoader(serviceClassLoader);
+                        clientHandler.onCompleted();
+                    } finally {
+                        Thread.currentThread().setContextClassLoader(oldClassLoader);
+                    }
+
                 }
 
                 private void checkInitialize(Request request) {