Repository: alibaba/fescar
Pull request #660 (2019-03-26 07:58:51)

fix: throw NPE when jdk-proxy has no target
<!-- Please make sure you have read and understood the contributing guidelines -->

### Ⅰ. Describe what this PR did
In the process of integrating spring-cloud-alibaba-stream-rocketmq with fescar, I throw a null pointer exception in the fescar-spring module's com.alibaba.fescar.spring.annotation.GlobalTransactionScanner. I found that com.alibaba was called. The .fescar.spring.util.SpringProxyUtils#findTargetClass method will recursively call target.getClass() when it gets the target object of the proxy class to generate a null pointer. The jdkProxy object generated by @EnableBinding has no target object, and there may be similar in the future. The scene, such as the proxy object generated by interface + annotation in Feign, also has no target object.
So I am modifying the com.alibaba.fescar.spring.util.SpringProxyUtils#findTargetClass part of the code:

`public static Class<?> findTargetClass(Object proxy) throws Exception {
        if (AopUtils.isAopProxy(proxy)) {
            AdvisedSupport advised = getAdvisedSupport(proxy);
            if(AopUtils.isJdkDynamicProxy(proxy)){
                TargetSource targetSource = advised.getTargetSource();
                return targetSource instanceof EmptyTargetSource ?  getFirstInterfaceByAdvised(advised): targetSource.getTarget().getClass();
            }
            Object target = advised.getTargetSource().getTarget();
            return findTargetClass(target);
        } else {
            return proxy.getClass();
        }
    }
    private static Class<?> getFirstInterfaceByAdvised(AdvisedSupport advised) {
        Class<?>[] interfaces = advised.getProxiedInterfaces();
        if(interfaces.length > 0){
            return interfaces[0];
        }else{
            throw new IllegalStateException("Find the jdk dynamic proxy class that does not implement the interface");
        }
    }`

When it is a jdk object, get its target object, if it can't get it, get the first interface it implements.
Ps. Perhaps you should think of a way to scan it to both (target object and interface)
Then my project is running normally.

### Ⅱ. Does this pull request fix one issue?
<!-- If that, add "fixes #xxx" below in the next line, for example, fixes #97. -->
Maybe I should mention an issue first.

### Ⅲ. Why don't you add test cases (unit test/integration test)? 


### Ⅳ. Describe how to verify it
NPE will be reported when integrating spring-cloud-stream and fescar

### Ⅴ. Special notes for reviews



codecov-io: # [Codecov](https://codecov.io/gh/alibaba/fescar/pull/660?src=pr&el=h1) Report
> Merging [#660](https://codecov.io/gh/alibaba/fescar/pull/660?src=pr&el=desc) into [develop](https://codecov.io/gh/alibaba/fescar/commit/e48c60e214323e51fc2ace6e7ffc9accedeafb21?src=pr&el=desc) will **decrease** coverage by `0.02%`.
> The diff coverage is `0%`.

[![Impacted file tree graph](https://codecov.io/gh/alibaba/fescar/pull/660/graphs/tree.svg?width=650&token=Hy9a6MCc4p&height=150&src=pr)](https://codecov.io/gh/alibaba/fescar/pull/660?src=pr&el=tree)

```diff
@@              Coverage Diff              @@
##             develop     #660      +/-   ##
=============================================
- Coverage      32.21%   32.19%   -0.03%     
  Complexity       876      876              
=============================================
  Files            226      226              
  Lines           8772     8779       +7     
  Branches        1055     1058       +3     
=============================================
  Hits            2826     2826              
- Misses          5627     5634       +7     
  Partials         319      319
```


| [Impacted Files](https://codecov.io/gh/alibaba/fescar/pull/660?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...m/alibaba/fescar/spring/util/SpringProxyUtils.java](https://codecov.io/gh/alibaba/fescar/pull/660/diff?src=pr&el=tree#diff-c3ByaW5nL3NyYy9tYWluL2phdmEvY29tL2FsaWJhYmEvZmVzY2FyL3NwcmluZy91dGlsL1NwcmluZ1Byb3h5VXRpbHMuamF2YQ==) | `4.65% <0%> (-0.91%)` | `2 <0> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/alibaba/fescar/pull/660?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/alibaba/fescar/pull/660?src=pr&el=footer). Last update [e48c60e...84b8652](https://codecov.io/gh/alibaba/fescar/pull/660?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

XCXCXCXCX: ![fescar bug](https://user-images.githubusercontent.com/32658071/54980574-19c3a980-4fe1-11e9-83df-54fc7595e993.png)

slievrly: @XCXCXCXCX thank you.
slievrly: @XCXCXCXCX why not use @GlobalTransactional on Controller, Controller call feign service?
XCXCXCXCX: For the situation you said, I have tested it separately before, and there is no problem when using fescar alone.
I am here to integrate spring-cloud-stream-rocketmq, @EnableBinding will create proxy objects for DemoSink and DemoSource, but fescar's GlobalTransactionScanner will scan all beans and determine if the bean should add fescar's interceptor, but the proxy object here is There is no target, so NPE is generated.
slievrly: > For the situation you said, I have tested it separately before, and there is no problem when using fescar alone.
> I am here to integrate spring-cloud-stream-rocketmq, @EnableBinding will create proxy objects for DemoSink and DemoSource, but fescar's GlobalTransactionScanner will scan all beans and determine if the bean should add fescar's interceptor, but the proxy object here is There is no target, so NPE is generated.

can you dingTalk ping me?

Patch:
diff --git a/README.md b/README.md
index 121f15ef7..4d8462818 100644
--- a/README.md
+++ b/README.md
@@ -69,7 +69,21 @@ For more details about principle and design, please go to [Fescar wiki page](htt
 
 <dependency>
     <groupId>com.alibaba.fescar</groupId>
-    <artifactId>fescar-all</artifactId>
+    <artifactId>fescar-spring</artifactId>
+    <version>${fescar.version}</version>
+</dependency>
+
+<!--dependency for Apache Dubbo-->
+<dependency>
+    <groupId>com.alibaba.fescar</groupId>
+    <artifactId>fescar-dubbo</artifactId>
+    <version>${fescar.version}</version>
+</dependency>
+
+<!--dependency for Alibaba Dubbo-->
+<dependency>
+    <groupId>com.alibaba.fescar</groupId>
+    <artifactId>fescar-dubbo-alibaba</artifactId>
     <version>${fescar.version}</version>
 </dependency>
 
@@ -95,9 +109,10 @@ Contributors are welcomed to join the FEATS project. Please check [CONTRIBUTING]
 ## Contact
 
 * [Twitter](): TBD. Follow along for latest Fescar news on Twitter.
-* Email Group:
-     * dev.fescar@gmail.com: Fescar developer discussion (APIs, feature design, etc).
 
+* Mailing list: 
+  * dev-fescar@googlegroups.com , for dev/user discussion. [subscribe](mailto:dev-fescar+subscribe@googlegroups.com), [unsubscribe](mailto:dev-fescar+unsubscribe@googlegroups.com), [archive](https://groups.google.com/forum/#!forum/dev-fescar)
+  
 **Dingtalk**
 
 <img src="https://upload-images.jianshu.io/upload_images/4420767-4e95b186a1a1bfba.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240"  height="300" width="300">
diff --git a/common/src/main/java/com/alibaba/fescar/common/thread/RejectedPolicys.java b/common/src/main/java/com/alibaba/fescar/common/thread/RejectedPolicys.java
new file mode 100644
index 000000000..bfaaf6722
--- /dev/null
+++ b/common/src/main/java/com/alibaba/fescar/common/thread/RejectedPolicys.java
@@ -0,0 +1,54 @@
+/*
+ *  Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ *  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 com.alibaba.fescar.common.thread;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * policys for RejectedExecutionHandler
+ *
+ * Created by guoyao on 2019/2/26.
+ */
+public final class RejectedPolicys {
+
+    /**
+     * when rejected happened ,add the new task and run the oldest task
+     *
+     * @return rejected execution handler
+     */
+    public static RejectedExecutionHandler runsOldestTaskPolicy() {
+        return new RejectedExecutionHandler() {
+            @Override
+            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+                if (executor.isShutdown()) {
+                    return;
+                }
+                BlockingQueue<Runnable> workQueue = executor.getQueue();
+                Runnable firstWork = workQueue.poll();
+                boolean newTaskAdd = workQueue.offer(r);
+                if (firstWork != null) {
+                    firstWork.run();
+                }
+                if (!newTaskAdd) {
+                    executor.execute(r);
+                }
+            }
+        };
+    }
+}
diff --git a/core/src/main/java/com/alibaba/fescar/core/rpc/netty/AbstractRpcRemotingServer.java b/core/src/main/java/com/alibaba/fescar/core/rpc/netty/AbstractRpcRemotingServer.java
index 2166db0d9..83d092471 100644
--- a/core/src/main/java/com/alibaba/fescar/core/rpc/netty/AbstractRpcRemotingServer.java
+++ b/core/src/main/java/com/alibaba/fescar/core/rpc/netty/AbstractRpcRemotingServer.java
@@ -18,6 +18,7 @@ package com.alibaba.fescar.core.rpc.netty;
 
 import java.net.InetSocketAddress;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 import com.alibaba.fescar.common.XID;
 import com.alibaba.fescar.common.thread.NamedThreadFactory;
@@ -150,6 +151,7 @@ public abstract class AbstractRpcRemotingServer extends AbstractRpcRemoting impl
             ChannelFuture future = this.serverBootstrap.bind(listenPort).sync();
             LOGGER.info("Server started ... ");
             RegistryFactory.getInstance().register(new InetSocketAddress(XID.getIpAddress(), XID.getPort()));
+            registerShutdownHook();
             future.channel().closeFuture().sync();
         } catch (Exception exx) {
             throw new RuntimeException(exx);
@@ -157,9 +159,24 @@ public abstract class AbstractRpcRemotingServer extends AbstractRpcRemoting impl
 
     }
 
+    /**
+     * Register vm shutdown hook to shutdown gracefully
+     * can't prevent kill -9
+     */
+    private void registerShutdownHook() {
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> shutdown()));
+    }
+
     @Override
     public void shutdown() {
         try {
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug("Shuting server down. ");
+            }
+            RegistryFactory.getInstance().unregister(new InetSocketAddress(XID.getIpAddress(), XID.getPort()));
+            //wait a few seconds for server transport
+            TimeUnit.SECONDS.sleep(nettyServerConfig.getServerShutdownWaitTime());
+
             this.eventLoopGroupBoss.shutdownGracefully();
             this.eventLoopGroupWorker.shutdownGracefully();
         } catch (Exception exx) {
diff --git a/core/src/main/java/com/alibaba/fescar/core/rpc/netty/NettyServerConfig.java b/core/src/main/java/com/alibaba/fescar/core/rpc/netty/NettyServerConfig.java
index 53a14e5a4..11d8e2d77 100644
--- a/core/src/main/java/com/alibaba/fescar/core/rpc/netty/NettyServerConfig.java
+++ b/core/src/main/java/com/alibaba/fescar/core/rpc/netty/NettyServerConfig.java
@@ -45,6 +45,12 @@ public class NettyServerConfig extends NettyBaseConfig {
     private static final String NIO_WORKER_THREAD_PREFIX = "NettyServerNIOWorker";
     private static final String DEFAULT_EXECUTOR_THREAD_PREFIX = "NettyServerBizHandler";
     private static final int DEFAULT_BOSS_THREAD_SIZE = 1;
+
+    /**
+     * Shutdown timeout default 1s
+     */
+    private static final int DEFAULT_SHUTDOWN_TIMEOUT_SEC = 1;
+
     /**
      * The Server channel clazz.
      */
@@ -296,4 +302,12 @@ public class NettyServerConfig extends NettyBaseConfig {
         return CONFIG.getInt("transport.thread-factory.boss-thread-size", DEFAULT_BOSS_THREAD_SIZE);
     }
 
+    /**
+     * Get the timeout seconds of shutdown.
+     *
+     * @return the int
+     */
+    public int getServerShutdownWaitTime() {
+        return CONFIG.getInt("transport.shutdown.wait", DEFAULT_SHUTDOWN_TIMEOUT_SEC);
+    }
 }
diff --git a/discovery/src/main/java/com/alibaba/fescar/discovery/registry/RedisRegistryServiceImpl.java b/discovery/src/main/java/com/alibaba/fescar/discovery/registry/RedisRegistryServiceImpl.java
index f76af35ad..6f122e3b8 100644
--- a/discovery/src/main/java/com/alibaba/fescar/discovery/registry/RedisRegistryServiceImpl.java
+++ b/discovery/src/main/java/com/alibaba/fescar/discovery/registry/RedisRegistryServiceImpl.java
@@ -68,7 +68,7 @@ public class RedisRegistryServiceImpl implements RegistryService<RedisListener>
         new NamedThreadFactory("RedisRegistryService", 1));
 
     private RedisRegistryServiceImpl() {
-        Configuration fescarConfig = ConfigurationFactory.getInstance();
+        Configuration fescarConfig = ConfigurationFactory.FILE_INSTANCE;
         this.clusterName = fescarConfig.getConfig(REDIS_FILEKEY_PREFIX + REGISTRY_CLUSTER_KEY, DEFAULT_CLUSTER);
         String password = fescarConfig.getConfig(getRedisPasswordFileKey());
         String serverAddr = fescarConfig.getConfig(getRedisAddrFileKey());
diff --git a/rm-datasource/src/main/java/com/alibaba/fescar/rm/datasource/exec/InsertExecutor.java b/rm-datasource/src/main/java/com/alibaba/fescar/rm/datasource/exec/InsertExecutor.java
index c059160af..8282cb61d 100644
--- a/rm-datasource/src/main/java/com/alibaba/fescar/rm/datasource/exec/InsertExecutor.java
+++ b/rm-datasource/src/main/java/com/alibaba/fescar/rm/datasource/exec/InsertExecutor.java
@@ -16,14 +16,6 @@
 
 package com.alibaba.fescar.rm.datasource.exec;
 
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
 import com.alibaba.fescar.common.exception.NotSupportYetException;
 import com.alibaba.fescar.common.exception.ShouldNeverHappenException;
 import com.alibaba.fescar.rm.datasource.PreparedStatementProxy;
@@ -31,17 +23,30 @@ import com.alibaba.fescar.rm.datasource.StatementProxy;
 import com.alibaba.fescar.rm.datasource.sql.SQLInsertRecognizer;
 import com.alibaba.fescar.rm.datasource.sql.SQLRecognizer;
 import com.alibaba.fescar.rm.datasource.sql.struct.ColumnMeta;
+import com.alibaba.fescar.rm.datasource.sql.struct.Null;
 import com.alibaba.fescar.rm.datasource.sql.struct.TableMeta;
 import com.alibaba.fescar.rm.datasource.sql.struct.TableRecords;
 
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
 /**
  * The type Insert executor.
  *
  * @param <T> the type parameter
  * @param <S> the type parameter
+ * @author yuanguoyao
+ * @date 2019-03-21 21:30:02
  */
 public class InsertExecutor<T, S extends Statement> extends AbstractDMLBaseExecutor<T, S> {
 
+    protected static final String ERR_SQL_STATE = "S1009";
+
     /**
      * Instantiates a new Insert executor.
      *
@@ -61,75 +66,89 @@ public class InsertExecutor<T, S extends Statement> extends AbstractDMLBaseExecu
 
     @Override
     protected TableRecords afterImage(TableRecords beforeImage) throws SQLException {
+        //Pk column exists or PK is just auto generated
+        List<Object> pkValues = containsPK() ? getPkValuesByColumn() : getPkValuesByAuto();
+
+        TableRecords afterImage = getTableRecords(pkValues);
+
+        if (afterImage == null) {
+            throw new SQLException("Failed to build after-image for insert");
+        }
+
+        return afterImage;
+    }
+
+    protected boolean containsPK() {
         SQLInsertRecognizer recogizier = (SQLInsertRecognizer)sqlRecognizer;
         List<String> insertColumns = recogizier.getInsertColumns();
         TableMeta tmeta = getTableMeta();
-        TableRecords afterImage = null;
-        if (tmeta.containsPK(insertColumns)) {
-            // insert values including PK
-            List<Object> pkValues = null;
-            String pk = tmeta.getPkName();
-            for (int paramIdx = 0; paramIdx < insertColumns.size(); paramIdx++) {
-                if (insertColumns.get(paramIdx).equalsIgnoreCase(pk)) {
-                    if (statementProxy instanceof PreparedStatementProxy) {
-                        pkValues = ((PreparedStatementProxy)statementProxy).getParamsByIndex(paramIdx);
-                    } else {
-                        List<List<Object>> insertRows = recogizier.getInsertRows();
-                        pkValues = new ArrayList<>(insertRows.size());
-                        for (List<Object> row : insertRows) {
-                            pkValues.add(row.get(paramIdx));
-                        }
-                    }
-                    break;
-                }
-            }
-            if (pkValues == null) {
-                throw new ShouldNeverHappenException();
-            }
-            afterImage = getTableRecords(pkValues);
-
-        } else {
-            // PK is just auto generated
-            Map<String, ColumnMeta> pkMetaMap = getTableMeta().getPrimaryKeyMap();
-            if (pkMetaMap.size() != 1) {
-                throw new NotSupportYetException();
-            }
-            ColumnMeta pkMeta = pkMetaMap.values().iterator().next();
-            if (!pkMeta.isAutoincrement()) {
-                throw new ShouldNeverHappenException();
-            }
+        return tmeta.containsPK(insertColumns);
+    }
 
-            ResultSet genKeys = null;
-            try {
-                genKeys = statementProxy.getTargetStatement().getGeneratedKeys();
-            } catch (SQLException e) {
-                // java.sql.SQLException: Generated keys not requested. You need to
-                // specify Statement.RETURN_GENERATED_KEYS to
-                // Statement.executeUpdate() or Connection.prepareStatement().
-                if ("S1009".equalsIgnoreCase(e.getSQLState())) {
-                    genKeys = statementProxy.getTargetStatement().executeQuery("SELECT LAST_INSERT_ID()");
+    protected List<Object> getPkValuesByColumn() throws SQLException {
+        // insert values including PK
+        SQLInsertRecognizer recogizier = (SQLInsertRecognizer)sqlRecognizer;
+        List<String> insertColumns = recogizier.getInsertColumns();
+        String pk = getTableMeta().getPkName();
+        List<Object> pkValues = null;
+        for (int paramIdx = 0; paramIdx < insertColumns.size(); paramIdx++) {
+            if (insertColumns.get(paramIdx).equalsIgnoreCase(pk)) {
+                if (statementProxy instanceof PreparedStatementProxy) {
+                    pkValues = ((PreparedStatementProxy)statementProxy).getParamsByIndex(paramIdx);
                 } else {
-                    throw e;
+                    List<List<Object>> insertRows = recogizier.getInsertRows();
+                    pkValues = new ArrayList<>(insertRows.size());
+                    for (List<Object> row : insertRows) {
+                        pkValues.add(row.get(paramIdx));
+                    }
                 }
+                break;
             }
-            List<Object> pkValues = new ArrayList<>();
-            while (genKeys.next()) {
-                Object v = genKeys.getObject(1);
-                pkValues.add(v);
-            }
+        }
+        if (pkValues == null) {
+            throw new ShouldNeverHappenException();
+        }
+        //pk auto generated while column exists and value is null
+        if (pkValues.size() == 1 && pkValues.get(0) instanceof Null) {
+            pkValues = getPkValuesByAuto();
+        }
+        return pkValues;
+    }
 
-            afterImage = getTableRecords(pkValues);
 
+    protected List<Object> getPkValuesByAuto() throws SQLException {
+        // PK is just auto generated
+        Map<String, ColumnMeta> pkMetaMap = getTableMeta().getPrimaryKeyMap();
+        if (pkMetaMap.size() != 1) {
+            throw new NotSupportYetException();
         }
-
-        if (afterImage == null) {
-            throw new SQLException("Failed to build after-image for insert");
+        ColumnMeta pkMeta = pkMetaMap.values().iterator().next();
+        if (!pkMeta.isAutoincrement()) {
+            throw new ShouldNeverHappenException();
         }
 
-        return afterImage;
+        ResultSet genKeys = null;
+        try {
+            genKeys = statementProxy.getTargetStatement().getGeneratedKeys();
+        } catch (SQLException e) {
+            // java.sql.SQLException: Generated keys not requested. You need to
+            // specify Statement.RETURN_GENERATED_KEYS to
+            // Statement.executeUpdate() or Connection.prepareStatement().
+            if (ERR_SQL_STATE.equalsIgnoreCase(e.getSQLState())) {
+                genKeys = statementProxy.getTargetStatement().executeQuery("SELECT LAST_INSERT_ID()");
+            } else {
+                throw e;
+            }
+        }
+        List<Object> pkValues = new ArrayList<>();
+        while (genKeys.next()) {
+            Object v = genKeys.getObject(1);
+            pkValues.add(v);
+        }
+        return pkValues;
     }
 
-    private TableRecords getTableRecords(List<Object> pkValues) throws SQLException {
+    protected TableRecords getTableRecords(List<Object> pkValues) throws SQLException {
         TableRecords afterImage;
         String pk = getTableMeta().getPkName();
         StringBuffer selectSQLAppender = new StringBuffer("SELECT * FROM " + getTableMeta().getTableName() + " WHERE ");
diff --git a/spring/src/main/java/com/alibaba/fescar/spring/annotation/GlobalTransactionalInterceptor.java b/spring/src/main/java/com/alibaba/fescar/spring/annotation/GlobalTransactionalInterceptor.java
index 5e3b856b0..17aff9dc5 100644
--- a/spring/src/main/java/com/alibaba/fescar/spring/annotation/GlobalTransactionalInterceptor.java
+++ b/spring/src/main/java/com/alibaba/fescar/spring/annotation/GlobalTransactionalInterceptor.java
@@ -33,6 +33,9 @@ import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.core.BridgeMethodResolver;
+import org.springframework.util.ClassUtils;
 
 /**
  * The type Global transactional interceptor.
@@ -60,12 +63,15 @@ public class GlobalTransactionalInterceptor implements MethodInterceptor {
 
     @Override
     public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
-        final GlobalTransactional globalTrxAnno = getAnnotation(methodInvocation.getMethod(),
-            GlobalTransactional.class);
-        final GlobalLock globalLockAnno = getAnnotation(methodInvocation.getMethod(), GlobalLock.class);
-        if (globalTrxAnno != null) {
-            return handleGlobalTransaction(methodInvocation, globalTrxAnno);
-        } else if (globalLockAnno != null) {
+        Class<?> targetClass = (methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null);
+        Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
+        final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
+
+        final GlobalTransactional globalTransactionalAnnotation = getAnnotation(method, GlobalTransactional.class);
+        final GlobalLock globalLockAnnotation = getAnnotation(method, GlobalLock.class);
+        if (globalTransactionalAnnotation != null) {
+            return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
+        } else if (globalLockAnnotation != null) {
             return handleGlobalLock(methodInvocation);
         } else {
             return methodInvocation.proceed();
diff --git a/spring/src/main/java/com/alibaba/fescar/spring/util/SpringProxyUtils.java b/spring/src/main/java/com/alibaba/fescar/spring/util/SpringProxyUtils.java
index 9459a38b5..05749ebe5 100644
--- a/spring/src/main/java/com/alibaba/fescar/spring/util/SpringProxyUtils.java
+++ b/spring/src/main/java/com/alibaba/fescar/spring/util/SpringProxyUtils.java
@@ -16,8 +16,10 @@
 
 package com.alibaba.fescar.spring.util;
 
+import org.springframework.aop.TargetSource;
 import org.springframework.aop.framework.AdvisedSupport;
 import org.springframework.aop.support.AopUtils;
+import org.springframework.aop.target.EmptyTargetSource;
 
 import java.lang.reflect.Field;
 import java.lang.reflect.Proxy;
@@ -39,6 +41,10 @@ public class SpringProxyUtils {
     public static Class<?> findTargetClass(Object proxy) throws Exception {
         if (AopUtils.isAopProxy(proxy)) {
             AdvisedSupport advised = getAdvisedSupport(proxy);
+            if(AopUtils.isJdkDynamicProxy(proxy)){
+                TargetSource targetSource = advised.getTargetSource();
+                return targetSource instanceof EmptyTargetSource ?  getFirstInterfaceByAdvised(advised): targetSource.getTarget().getClass();
+            }
             Object target = advised.getTargetSource().getTarget();
             return findTargetClass(target);
         } else {
@@ -46,6 +52,15 @@ public class SpringProxyUtils {
         }
     }
 
+    private static Class<?> getFirstInterfaceByAdvised(AdvisedSupport advised) {
+        Class<?>[] interfaces = advised.getProxiedInterfaces();
+        if(interfaces.length > 0){
+            return interfaces[0];
+        }else{
+            throw new IllegalStateException("Find the jdk dynamic proxy class that does not implement the interface");
+        }
+    }
+
     /**
      * Gets advised support.
      *


Test patch:
diff --git a/common/src/test/java/com/alibaba/fescar/common/thread/RejectedPolicysTest.java b/common/src/test/java/com/alibaba/fescar/common/thread/RejectedPolicysTest.java
new file mode 100644
index 000000000..64e90e715
--- /dev/null
+++ b/common/src/test/java/com/alibaba/fescar/common/thread/RejectedPolicysTest.java
@@ -0,0 +1,97 @@
+/*
+ *  Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ *  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 com.alibaba.fescar.common.thread;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Created by guoyao on 2019/2/26.
+ */
+public class RejectedPolicysTest {
+
+    private final int DEFAULT_CORE_POOL_SIZE = 1;
+    private final int DEFAULT_KEEP_ALIVE_TIME = 10;
+    private final int MAX_QUEUE_SIZE = 1;
+
+    /**
+     * Test runs oldest task policy.
+     *
+     * @throws Exception the exception
+     */
+    @Test
+    public void testRunsOldestTaskPolicy() throws Exception {
+        AtomicInteger atomicInteger = new AtomicInteger();
+        ThreadPoolExecutor poolExecutor =
+            new ThreadPoolExecutor(DEFAULT_CORE_POOL_SIZE, DEFAULT_CORE_POOL_SIZE, DEFAULT_KEEP_ALIVE_TIME,
+                TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue(MAX_QUEUE_SIZE),
+                new NamedThreadFactory("OldestRunsPolicy", DEFAULT_CORE_POOL_SIZE),
+                RejectedPolicys.runsOldestTaskPolicy());
+        CountDownLatch downLatch1 = new CountDownLatch(1);
+        CountDownLatch downLatch2 = new CountDownLatch(1);
+        CountDownLatch downLatch3 = new CountDownLatch(1);
+        //task1
+        poolExecutor.execute(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    //wait the oldest task of queue count down
+                    downLatch1.await();
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+                atomicInteger.getAndAdd(1);
+            }
+        });
+        assertThat(atomicInteger.get()).isEqualTo(0);
+        //task2
+        poolExecutor.execute(new Runnable() {
+            @Override
+            public void run() {
+                // run second
+                atomicInteger.getAndAdd(2);
+            }
+        });
+        //task3
+        poolExecutor.execute(new Runnable() {
+            @Override
+            public void run() {
+                downLatch2.countDown();
+                //task3 run
+                atomicInteger.getAndAdd(3);
+                downLatch3.countDown();
+            }
+        });
+        //only the task2 run which is the oldest task of queue
+        assertThat(atomicInteger.get()).isEqualTo(2);
+        downLatch1.countDown();
+        downLatch2.await();
+        //wait task3 run +3
+        downLatch3.await();
+        //run task3
+        assertThat(atomicInteger.get()).isEqualTo(6);
+
+    }
+}
diff --git a/rm-datasource/src/test/java/com/alibaba/fescar/rm/datasource/exec/InsertExecutorTest.java b/rm-datasource/src/test/java/com/alibaba/fescar/rm/datasource/exec/InsertExecutorTest.java
new file mode 100644
index 000000000..633eea09c
--- /dev/null
+++ b/rm-datasource/src/test/java/com/alibaba/fescar/rm/datasource/exec/InsertExecutorTest.java
@@ -0,0 +1,275 @@
+/*
+ *  Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ *  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 com.alibaba.fescar.rm.datasource.exec;
+
+import com.alibaba.fescar.common.exception.NotSupportYetException;
+import com.alibaba.fescar.common.exception.ShouldNeverHappenException;
+import com.alibaba.fescar.rm.datasource.PreparedStatementProxy;
+import com.alibaba.fescar.rm.datasource.sql.SQLInsertRecognizer;
+import com.alibaba.fescar.rm.datasource.sql.struct.*;
+import org.assertj.core.api.Assertions;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.anyString;
+
+/**
+ * @author guoyao
+ * @date 2019/3/21
+ */
+public class InsertExecutorTest {
+
+    private static final String ID_COLUMN = "id";
+    private static final String USER_ID_COLUMN = "user_id";
+    private static final String USER_NAME_COLUMN = "user_name";
+    private static final Integer PK_VALUE = 100;
+
+    private PreparedStatementProxy statementProxy;
+
+    private SQLInsertRecognizer sqlInsertRecognizer;
+
+    private StatementCallback statementCallback;
+
+    private TableMeta tableMeta;
+
+    private InsertExecutor insertExecutor;
+
+    @Before
+    public void init() {
+        statementProxy = mock(PreparedStatementProxy.class);
+        statementCallback = mock(StatementCallback.class);
+        sqlInsertRecognizer = mock(SQLInsertRecognizer.class);
+        tableMeta =  mock(TableMeta.class);
+        insertExecutor = spy(new InsertExecutor(statementProxy, statementCallback, sqlInsertRecognizer));
+    }
+
+    @Test
+    public void testBeforeImage() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        TableRecords tableRecords = insertExecutor.beforeImage();
+        Assertions.assertThat(tableRecords.getRows()).isEmpty();
+        Assertions.assertThat(tableRecords.size()).isEqualTo(0);
+        try {
+            tableRecords.add(new Row());
+        } catch (Exception e) {
+            Assertions.assertThat(e instanceof UnsupportedOperationException).isTrue();
+        }
+        try {
+            tableRecords.getTableMeta();
+        } catch (Exception e) {
+            Assertions.assertThat(e instanceof UnsupportedOperationException).isTrue();
+        }
+    }
+
+    @Test
+    public void testAfterImage_ByColumn() throws SQLException {
+        doReturn(true).when(insertExecutor).containsPK();
+        List<Object> pkValues = new ArrayList<>();
+        pkValues.add(PK_VALUE);
+        doReturn(pkValues).when(insertExecutor).getPkValuesByColumn();
+        TableRecords tableRecords = new TableRecords();
+        doReturn(tableRecords).when(insertExecutor).getTableRecords(pkValues);
+        TableRecords resultTableRecords = insertExecutor.afterImage(new TableRecords());
+        Assertions.assertThat(resultTableRecords).isEqualTo(tableRecords);
+    }
+
+    @Test
+    public void testAfterImage_ByAuto() throws SQLException {
+        doReturn(false).when(insertExecutor).containsPK();
+        List<Object> pkValues = new ArrayList<>();
+        pkValues.add(PK_VALUE);
+        doReturn(pkValues).when(insertExecutor).getPkValuesByAuto();
+        TableRecords tableRecords = new TableRecords();
+        doReturn(tableRecords).when(insertExecutor).getTableRecords(pkValues);
+        TableRecords resultTableRecords = insertExecutor.afterImage(new TableRecords());
+        Assertions.assertThat(resultTableRecords).isEqualTo(tableRecords);
+    }
+
+    @Test(expected = SQLException.class)
+    public void testAfterImage_Exception() throws SQLException {
+        doReturn(false).when(insertExecutor).containsPK();
+        List<Object> pkValues = new ArrayList<>();
+        pkValues.add(PK_VALUE);
+        doReturn(pkValues).when(insertExecutor).getPkValuesByAuto();
+        doReturn(null).when(insertExecutor).getTableRecords(pkValues);
+        insertExecutor.afterImage(new TableRecords());
+    }
+
+    @Test
+    public void testContainsPK() {
+        List<String> insertColumns = mockInsertColumns();
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        when(tableMeta.containsPK(insertColumns)).thenReturn(true);
+        Assertions.assertThat(insertExecutor.containsPK()).isTrue();
+        when(tableMeta.containsPK(insertColumns)).thenReturn(false);
+        Assertions.assertThat(insertExecutor.containsPK()).isFalse();
+    }
+
+    @Test
+    public void testGetPkValuesByColumn() throws SQLException {
+        mockInsertColumns();
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        when(tableMeta.getPkName()).thenReturn(ID_COLUMN);
+        List<Object> pkValues = new ArrayList<>();
+        pkValues.add(PK_VALUE);
+        when(statementProxy.getParamsByIndex(0)).thenReturn(pkValues);
+        List pkValuesByColumn=insertExecutor.getPkValuesByColumn();
+        Assertions.assertThat(pkValuesByColumn).isEqualTo(pkValues);
+    }
+
+    @Test(expected = ShouldNeverHappenException.class)
+    public void testGetPkValuesByColumn_Exception() throws SQLException {
+        mockInsertColumns();
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        when(tableMeta.getPkName()).thenReturn(ID_COLUMN);
+        when(statementProxy.getParamsByIndex(0)).thenReturn(null);
+        insertExecutor.getPkValuesByColumn();
+    }
+
+    @Test
+    public void testGetPkValuesByColumn_PkValue_Null() throws SQLException {
+        mockInsertColumns();
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        when(tableMeta.getPkName()).thenReturn(ID_COLUMN);
+        List<Object> pkValuesNull = new ArrayList<>();
+        pkValuesNull.add(Null.get());
+        when(statementProxy.getParamsByIndex(0)).thenReturn(pkValuesNull);
+        List<Object> pkValuesAuto = new ArrayList<>();
+        pkValuesAuto.add(PK_VALUE);
+        //mock getPkValuesByAuto
+        doReturn(pkValuesAuto).when(insertExecutor).getPkValuesByAuto();
+        List pkValuesByColumn = insertExecutor.getPkValuesByColumn();
+        //pk value = Null so getPkValuesByAuto
+        verify(insertExecutor).getPkValuesByAuto();
+        Assertions.assertThat(pkValuesByColumn).isEqualTo(pkValuesAuto);
+    }
+
+    @Test(expected = NotSupportYetException.class)
+    public void testGetPkValuesByAuto_NotSupportYetException() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        Map<String, ColumnMeta> columnMetaMap = new HashMap<>();
+        columnMetaMap.put(ID_COLUMN, new ColumnMeta());
+        columnMetaMap.put(USER_ID_COLUMN, new ColumnMeta());
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(columnMetaMap);
+        insertExecutor.getPkValuesByAuto();
+    }
+
+    @Test(expected = ShouldNeverHappenException.class)
+    public void testGetPkValuesByAuto_ShouldNeverHappenException() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        Map<String, ColumnMeta> columnMetaMap = new HashMap<>();
+        ColumnMeta columnMeta = mock(ColumnMeta.class);
+        columnMetaMap.put(ID_COLUMN, columnMeta);
+        when(columnMeta.isAutoincrement()).thenReturn(false);
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(columnMetaMap);
+        insertExecutor.getPkValuesByAuto();
+    }
+
+    @Test(expected = SQLException.class)
+    public void testGetPkValuesByAuto_SQLException() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        ColumnMeta columnMeta = mock(ColumnMeta.class);
+        Map<String, ColumnMeta> columnMetaMap = new HashMap<>();
+        columnMetaMap.put(ID_COLUMN, columnMeta);
+        when(columnMeta.isAutoincrement()).thenReturn(true);
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(columnMetaMap);
+        PreparedStatement preparedStatement = mock(PreparedStatement.class);
+        when(statementProxy.getTargetStatement()).thenReturn(preparedStatement);
+        when(preparedStatement.getGeneratedKeys()).thenThrow(new SQLException());
+        insertExecutor.getPkValuesByAuto();
+    }
+
+    @Test
+    public void testGetPkValuesByAuto_GeneratedKeys_NoResult() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        ColumnMeta columnMeta = mock(ColumnMeta.class);
+        Map<String, ColumnMeta> columnMetaMap = new HashMap<>();
+        columnMetaMap.put(ID_COLUMN, columnMeta);
+        when(columnMeta.isAutoincrement()).thenReturn(true);
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(columnMetaMap);
+        PreparedStatement preparedStatement = mock(PreparedStatement.class);
+        when(statementProxy.getTargetStatement()).thenReturn(preparedStatement);
+        ResultSet resultSet = mock(ResultSet.class);
+        when(preparedStatement.getGeneratedKeys()).thenReturn(resultSet);
+        when(resultSet.next()).thenReturn(false);
+        when(resultSet.getObject(1)).thenReturn(PK_VALUE);
+        List pkValuesByAuto = insertExecutor.getPkValuesByAuto();
+        Assertions.assertThat(pkValuesByAuto).isEmpty();
+    }
+
+    @Test
+    public void testGetPkValuesByAuto_GeneratedKeys_HasResult() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        ColumnMeta columnMeta = mock(ColumnMeta.class);
+        Map<String, ColumnMeta> columnMetaMap = new HashMap<>();
+        columnMetaMap.put(ID_COLUMN, columnMeta);
+        when(columnMeta.isAutoincrement()).thenReturn(true);
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(columnMetaMap);
+        PreparedStatement preparedStatement = mock(PreparedStatement.class);
+        when(statementProxy.getTargetStatement()).thenReturn(preparedStatement);
+        ResultSet resultSet = mock(ResultSet.class);
+        when(preparedStatement.getGeneratedKeys()).thenReturn(resultSet);
+        when(resultSet.next()).thenReturn(true).thenReturn(false);
+        when(resultSet.getObject(1)).thenReturn(PK_VALUE);
+        List<Object> pkValues = new ArrayList<>();
+        pkValues.add(PK_VALUE);
+        List pkValuesByAuto = insertExecutor.getPkValuesByAuto();
+        Assertions.assertThat(pkValuesByAuto).isEqualTo(pkValues);
+    }
+
+    @Test
+    public void testGetPkValuesByAuto_ExecuteQuery_HasResult() throws SQLException {
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        ColumnMeta columnMeta = mock(ColumnMeta.class);
+        Map<String, ColumnMeta> columnMetaMap = new HashMap<>();
+        columnMetaMap.put(ID_COLUMN, columnMeta);
+        when(columnMeta.isAutoincrement()).thenReturn(true);
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(columnMetaMap);
+        PreparedStatement preparedStatement = mock(PreparedStatement.class);
+        when(statementProxy.getTargetStatement()).thenReturn(preparedStatement);
+        when(preparedStatement.getGeneratedKeys()).thenThrow(new SQLException("", InsertExecutor.ERR_SQL_STATE));
+        ResultSet resultSet = mock(ResultSet.class);
+        when(preparedStatement.executeQuery(anyString())).thenReturn(resultSet);
+        when(resultSet.next()).thenReturn(true).thenReturn(false);
+        when(resultSet.getObject(1)).thenReturn(PK_VALUE);
+        List<Object> pkValues = new ArrayList<>();
+        pkValues.add(PK_VALUE);
+        List pkValuesByAuto = insertExecutor.getPkValuesByAuto();
+        Assertions.assertThat(pkValuesByAuto).isEqualTo(pkValues);
+    }
+
+    private List<String> mockInsertColumns() {
+        List<String> columns = new ArrayList<>();
+        columns.add(ID_COLUMN);
+        columns.add(USER_ID_COLUMN);
+        columns.add(USER_NAME_COLUMN);
+        when(sqlInsertRecognizer.getInsertColumns()).thenReturn(columns);
+        return columns;
+    }
+}