fix heartbeat deadlock inner 1372 (#2867)

# Conflicts:
#	src/main/java/com/actiontech/dble/net/NIOSocketWR.java

src/main/java/com/actiontech/dble/backend/mysql/nio/MySQLConnectionAuthenticator.java

@@ -43,6 +43,10 @@ public class MySQLConnectionAuthenticator implements NIOHandler {
             listener.connectionError(e, c);
         }
     }
+    // only for test
+    public ResponseHandler getListener() {
+        return listener;
+    }
 
     @Override
     public void handle(byte[] data) {

src/main/java/com/actiontech/dble/btrace/provider/ClusterDelayProvider.java

@@ -128,4 +128,9 @@ public final class ClusterDelayProvider {
 
     }
 
+    //only for test need delete
+    public static void heartbeat() {
+
+    }
+
 }

src/main/java/com/actiontech/dble/net/AbstractConnection.java

@@ -285,6 +285,7 @@ public abstract class AbstractConnection implements NIOConnection {
         if (got < 0) {
             if (this instanceof MySQLConnection) {
                 ((MySQLConnection) this).closeInner("stream closed");
+                ((MySQLConnection) this).onConnectFailed(new IOException("stream closed"));
             } else {
                 this.close("stream closed");
             }

src/main/java/com/actiontech/dble/net/NIOReactor.java

@@ -123,10 +123,16 @@ public final class NIOReactor {
                                 con.asyncRead();
                             } catch (IOException e) {
                                 con.close("program err:" + e.toString());
+                                if (con instanceof MySQLConnection) {
+                                    ((MySQLConnection) con).onConnectFailed(e);
+                                }
                                 continue;
                             } catch (Exception e) {
                                 LOGGER.warn("caught err:", e);
                                 con.close("program err:" + e.toString());
+                                if (con instanceof MySQLConnection) {
+                                    ((MySQLConnection) con).onConnectFailed(e);
+                                }
                                 continue;
                             }
                         }

src/main/java/com/actiontech/dble/net/NIOSocketWR.java

@@ -6,6 +6,9 @@
 package com.actiontech.dble.net;
 
 import com.actiontech.dble.backend.mysql.nio.MySQLConnection;
+import com.actiontech.dble.backend.mysql.nio.MySQLConnectionAuthenticator;
+import com.actiontech.dble.backend.mysql.nio.handler.NewConnectionRespHandler;
+import com.actiontech.dble.btrace.provider.ClusterDelayProvider;
 import com.actiontech.dble.util.TimeUtil;
 
 import java.io.IOException;
@@ -242,6 +245,11 @@ public class NIOSocketWR extends SocketWR {
 
     @Override
     public void asyncRead() throws IOException {
+        // only for test
+        if (con.getHandler() instanceof MySQLConnectionAuthenticator && ((MySQLConnectionAuthenticator) con.getHandler()).getListener() instanceof NewConnectionRespHandler) {
+            ClusterDelayProvider.heartbeat();
+        }
+
         ByteBuffer theBuffer = con.readBuffer;
         if (theBuffer == null) {
             theBuffer = con.processor.getBufferPool().allocate(con.processor.getBufferPool().getChunkSize());