Resume the realization of KillConnection (#2062)

* Resume the realization of KillConnection

* Remove duplicate codes

* Remove duplicate codes

src/main/java/com/actiontech/dble/net/service/FrontEndService.java

@@ -12,4 +12,6 @@ public interface FrontEndService {
     UserName getUser();
 
     String getExecuteSql();
+
+    void killAndClose(String reason);
 }

src/main/java/com/actiontech/dble/server/NonBlockingSession.java

@@ -876,7 +876,7 @@ public class NonBlockingSession extends Session {
         }
     }
 
-    protected void kill() {
+    public void kill() {
         AtomicInteger count = new AtomicInteger(0);
         Map<RouteResultsetNode, BackendConnection> toKilled = new HashMap<>();
 
@@ -972,13 +972,6 @@ public class NonBlockingSession extends Session {
             if (!isSuccess) {
                 LOGGER.warn("DDL execute failed or Session closed, " +
                         "Schema[" + rrs.getSchema() + "],SQL[" + sql + "]" + (errInfo != null ? "errorInfo:" + errInfo : ""));
-            } else {
-                DDLTraceManager.getInstance().updateDDLStatus(DDLTraceInfo.DDLStage.META_UPDATE, shardingService);
-            }
-
-            if (rrs.isOnline()) {
-                LOGGER.info("online ddl skip updating meta and cluster notify, Schema[" + rrs.getSchema() + "],SQL[" + sql + "]" + (errInfo != null ? "errorInfo:" + errInfo : ""));
-                return true;
             }
             DDLTraceManager.getInstance().updateDDLStatus(DDLTraceInfo.DDLStage.META_UPDATE, shardingService);
             return ProxyMeta.getInstance().getTmManager().updateMetaData(rrs.getSchema(), rrs.getTable(), sql, isSuccess, rrs.getDdlType());

src/main/java/com/actiontech/dble/services/manager/ManagerService.java

@@ -107,6 +107,11 @@ public class ManagerService extends MySQLBasedService implements FrontEndService
         return executeSql;
     }
 
+    @Override
+    public void killAndClose(String reason) {
+        connection.close(reason);
+    }
+
 
     public String getCommand(byte[] data, CharsetNames charsetName) throws UnsupportedEncodingException {
         String sql = null;

src/main/java/com/actiontech/dble/services/manager/response/KillConnection.java

@@ -33,8 +33,7 @@ public final class KillConnection {
             for (FrontendConnection c : list) {
                 StringBuilder s = new StringBuilder();
                 LOGGER.info(s.append(c).append("killed by manager").toString());
-                //todo frontend kill and close
-                //c.killAndClose("kill by manager");
+                c.getFrontEndService().killAndClose("kill by manager");
                 count++;
             }
         }

src/main/java/com/actiontech/dble/services/mysqlsharding/ShardingService.java

@@ -739,6 +739,15 @@ public class ShardingService extends MySQLBasedService implements FrontEndServic
         return executeSql;
     }
 
+    @Override
+    public void killAndClose(String reason) {
+        connection.close(reason);
+        if (!isTxStart() || session.getTransactionManager().getXAStage() == null) {
+            //not a xa transaction ,close it
+            session.kill();
+        }
+    }
+
     public void setExecuteSql(String executeSql) {
         this.executeSql = executeSql;
     }