Merge remote-tracking branch 'dubbo_rem/master'

Author: lovepoem

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/SerializationException.java

@@ -0,0 +1,79 @@
+/*
+ * 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.remoting;
+
+import java.net.InetSocketAddress;
+
+/**
+ * SerializationException. (API, Prototype, ThreadSafe)
+ *
+ * @export
+ * @see org.apache.dubbo.remoting.exchange.support.DefaultFuture#get()
+ */
+public class SerializationException extends Exception {
+
+    private static final long serialVersionUID = -3160452149606778709L;
+
+    private InetSocketAddress localAddress;
+
+    private InetSocketAddress remoteAddress;
+
+    public SerializationException(Channel channel, String msg) {
+        this(channel == null ? null : channel.getLocalAddress(), channel == null ? null : channel.getRemoteAddress(),
+                msg);
+    }
+
+    public SerializationException(InetSocketAddress localAddress, InetSocketAddress remoteAddress, String message) {
+        super(message);
+
+        this.localAddress = localAddress;
+        this.remoteAddress = remoteAddress;
+    }
+
+    public SerializationException(Channel channel, Throwable cause) {
+        this(channel == null ? null : channel.getLocalAddress(), channel == null ? null : channel.getRemoteAddress(),
+                cause);
+    }
+
+    public SerializationException(InetSocketAddress localAddress, InetSocketAddress remoteAddress, Throwable cause) {
+        super(cause);
+
+        this.localAddress = localAddress;
+        this.remoteAddress = remoteAddress;
+    }
+
+    public SerializationException(Channel channel, String message, Throwable cause) {
+        this(channel == null ? null : channel.getLocalAddress(), channel == null ? null : channel.getRemoteAddress(),
+                message, cause);
+    }
+
+    public SerializationException(InetSocketAddress localAddress, InetSocketAddress remoteAddress, String message,
+                                  Throwable cause) {
+        super(message, cause);
+
+        this.localAddress = localAddress;
+        this.remoteAddress = remoteAddress;
+    }
+
+    public InetSocketAddress getLocalAddress() {
+        return localAddress;
+    }
+
+    public InetSocketAddress getRemoteAddress() {
+        return remoteAddress;
+    }
+}

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/Response.java

@@ -28,6 +28,11 @@ public class Response {
      */
     public static final byte OK = 20;
 
+    /**
+     * serialization error
+     */
+    public static final byte SERIALIZATION_ERROR = 25;
+
     /**
      * client side timeout.
      */

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/codec/ExchangeCodec.java

@@ -347,7 +347,7 @@ protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response re
             // send error message to Consumer, otherwise, Consumer will wait till timeout.
             if (!res.isEvent() && res.getStatus() != Response.BAD_RESPONSE) {
                 Response r = new Response(res.getId(), res.getVersion());
-                r.setStatus(Response.BAD_RESPONSE);
+                r.setStatus(Response.SERIALIZATION_ERROR);
 
                 if (t instanceof ExceedPayloadLimitException) {
                     logger.warn(t.getMessage(), t);

dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java

@@ -26,6 +26,7 @@
 import org.apache.dubbo.common.utils.NamedThreadFactory;
 import org.apache.dubbo.remoting.Channel;
 import org.apache.dubbo.remoting.RemotingException;
+import org.apache.dubbo.remoting.SerializationException;
 import org.apache.dubbo.remoting.TimeoutException;
 import org.apache.dubbo.remoting.exchange.Request;
 import org.apache.dubbo.remoting.exchange.Response;
@@ -203,7 +204,9 @@ private void doReceived(Response res) {
             this.complete(res.getResult());
         } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
             this.completeExceptionally(new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage()));
-        } else {
+        } else if(res.getStatus() == Response.SERIALIZATION_ERROR){
+            this.completeExceptionally(new SerializationException(channel, res.getErrorMessage()));
+        }else {
             this.completeExceptionally(new RemotingException(channel, res.getErrorMessage()));
         }
 

dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/codec/ExchangeCodecTest.java

@@ -498,7 +498,7 @@ public void testMessageLengthExceedPayloadLimitWhenEncode() throws Exception {
         codec.encode(channel, encodeBuffer, response);
         Assertions.assertTrue(channel.getReceivedMessage() instanceof Response);
         Response receiveMessage = (Response) channel.getReceivedMessage();
-        Assertions.assertEquals(Response.BAD_RESPONSE, receiveMessage.getStatus());
+        Assertions.assertEquals(Response.SERIALIZATION_ERROR, receiveMessage.getStatus());
         Assertions.assertTrue(receiveMessage.getErrorMessage().contains("Data length too large: "));
     }
 }

dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/transport/codec/DeprecatedExchangeCodec.java

@@ -279,7 +279,7 @@ protected void encodeResponse(Channel channel, OutputStream os, Response res) th
                     logger.warn("Fail to encode response: " + res + ", send bad_response info instead, cause: " + t.getMessage(), t);
 
                     Response r = new Response(res.getId(), res.getVersion());
-                    r.setStatus(Response.BAD_RESPONSE);
+                    r.setStatus(Response.SERIALIZATION_ERROR);
                     r.setErrorMessage("Failed to send response: " + res + ", cause: " + StringUtils.toString(t));
                     channel.send(r);
 

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyClientHandler.java

@@ -16,6 +16,7 @@
  */
 package org.apache.dubbo.remoting.transport.netty4;
 
+import io.netty.handler.codec.EncoderException;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.Version;
 import org.apache.dubbo.common.logger.Logger;
@@ -154,7 +155,11 @@ public void handshakeCompleted(SslHandlerInitializer.HandshakeCompletionEvent ev
      */
     private static Response buildErrorResponse(Request request, Throwable t) {
         Response response = new Response(request.getId(), request.getVersion());
-        response.setStatus(Response.BAD_REQUEST);
+        if(t instanceof EncoderException){
+            response.setStatus(Response.SERIALIZATION_ERROR);
+        }else{
+            response.setStatus(Response.BAD_REQUEST);
+        }
         response.setErrorMessage(StringUtils.toString(t));
         return response;
     }

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/AsyncToSyncInvoker.java

@@ -18,6 +18,7 @@
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.remoting.RemotingException;
+import org.apache.dubbo.remoting.SerializationException;
 import org.apache.dubbo.remoting.TimeoutException;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.InvokeMode;
@@ -69,7 +70,10 @@ public Result invoke(Invocation invocation) throws RpcException {
             if (t instanceof TimeoutException) {
                 throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " +
                         invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
-            } else if (t instanceof RemotingException) {
+            } else if(t instanceof SerializationException){
+                throw new RpcException(RpcException.SERIALIZATION_EXCEPTION, "Failed to invoke remote method: " +
+                        invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
+            }else if (t instanceof RemotingException) {
                 throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " +
                         invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
             } else {