optimize: optimize decode exception (apache#3895)

dongYES · Jul 23, 2021 · ba136f1 · ba136f1
1 parent 3a9f9d0
commit ba136f1
Show file tree

Hide file tree

Showing 5 changed files with 56 additions and 18 deletions.
diff --git a/changes/1.5.0.md b/changes/1.5.0.md
@@ -77,6 +77,7 @@ Seata 是一款开源的分布式事务解决方案，提供高性能和简单
   - [[#3847](https://github.com/seata/seata/pull/3847)] 优化ConcurrentHashMap.newKeySet替换ConcurrentSet
   - [[#3849](https://github.com/seata/seata/pull/3849)] 优化字符串拼接
   - [[#3699](https://github.com/seata/seata/pull/3699)] 优化 redis mock测试
+  - [[#3895](https://github.com/seata/seata/pull/3895)] 优化解码异常
 
 ### test：
 

diff --git a/changes/en-us/1.5.0.md b/changes/en-us/1.5.0.md
@@ -77,6 +77,7 @@
   - [[#3847](https://github.com/seata/seata/pull/3847)] optimize ConcurrentHashMap.newKeySet replace ConcurrentSet
   - [[#3849](https://github.com/seata/seata/pull/3849)] optimize string concat
   - [[#3699](https://github.com/seata/seata/pull/3699)] optimize redis mock test
+  - [[#3895](https://github.com/seata/seata/pull/3895)] optimize decode exception
 
 
 

diff --git a/core/src/main/java/io/seata/core/exception/DecodeException.java b/core/src/main/java/io/seata/core/exception/DecodeException.java
@@ -0,0 +1,27 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed 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 io.seata.core.exception;
+
+/**
+ * @author slievrly
+ */
+public class DecodeException extends Exception {
+
+    public DecodeException(Throwable throwable) {
+        super(throwable);
+    }
+}
diff --git a/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingServer.java b/core/src/main/java/io/seata/core/rpc/netty/AbstractNettyRemotingServer.java
@@ -15,11 +15,16 @@
  */
 package io.seata.core.rpc.netty;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeoutException;
+
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.DecoderException;
 import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 import io.seata.common.util.NetUtil;
@@ -33,10 +38,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeoutException;
-
 /**
  * The type abstract remoting server.
  *
@@ -229,11 +230,15 @@ private void handleDisconnect(ChannelHandlerContext ctx) {
          */
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-            if (LOGGER.isInfoEnabled()) {
-                LOGGER.info("channel exx:" + cause.getMessage() + ",channel:" + ctx.channel());
+            try {
+                if (cause instanceof DecoderException && null == ChannelManager.getContextFromIdentified(ctx.channel())) {
+                    return;
+                }
+                LOGGER.error("exceptionCaught:{}, channel:{}", cause.getMessage(), ctx.channel());
+                super.exceptionCaught(ctx, cause);
+            } finally {
+                ChannelManager.releaseRpcContext(ctx.channel());
             }
-            ChannelManager.releaseRpcContext(ctx.channel());
-            super.exceptionCaught(ctx, cause);
         }
 
         /**

diff --git a/core/src/main/java/io/seata/core/rpc/netty/v1/ProtocolV1Decoder.java b/core/src/main/java/io/seata/core/rpc/netty/v1/ProtocolV1Decoder.java
@@ -19,6 +19,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import io.seata.common.loader.EnhancedServiceLoader;
+import io.seata.core.exception.DecodeException;
 import io.seata.core.serializer.Serializer;
 import io.seata.core.compressor.Compressor;
 import io.seata.core.compressor.CompressorFactory;
@@ -80,17 +81,20 @@ int lengthFieldLength,  FullLength is int(4B). so values is 4
 
     @Override
     protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
-        Object decoded = super.decode(ctx, in);
-        if (decoded instanceof ByteBuf) {
-            ByteBuf frame = (ByteBuf) decoded;
-            try {
-                return decodeFrame(frame);
-            } catch (Exception e) {
-                LOGGER.error("Decode frame error!", e);
-                throw e;
-            } finally {
-                frame.release();
+        Object decoded;
+        try {
+            decoded = super.decode(ctx, in);
+            if (decoded instanceof ByteBuf) {
+                ByteBuf frame = (ByteBuf)decoded;
+                try {
+                    return decodeFrame(frame);
+                } finally {
+                    frame.release();
+                }
             }
+        } catch (Exception exx) {
+            LOGGER.error("Decode frame error, cause: {}", exx.getMessage());
+            throw new DecodeException(exx);
         }
         return decoded;
     }