1.repair front end connection leaks

2.improve monitor
3.fix replica monitor occurs null exception

common/src/main/java/io/mycat/beans/mysql/packet/ErrorPacketImpl.java

@@ -19,6 +19,8 @@
 
 import io.mycat.config.MySQLServerCapabilityFlags;
 
+import java.nio.charset.StandardCharsets;
+
 /**
  * https://mariadb.com/kb/en/library/err_packet/
  *
@@ -31,7 +33,7 @@ public class ErrorPacketImpl implements ErrorPacket {
     private int progress;
     private byte[] progress_info;
     private byte mark = ' ';
-    private byte[] sqlState = DEFAULT_SQLSTATE;
+    private byte[] sqlState = "".getBytes(StandardCharsets.UTF_8);
     private String message;
 
     public int getErrorCode() {
@@ -51,8 +53,8 @@ public void writePayload(MySQLPayloadWriteView buffer, int serverCapabilities) {
             buffer.writeFixInt(3, progress);
             buffer.writeLenencString(progress_info);
         } else if (MySQLServerCapabilityFlags.isClientProtocol41(serverCapabilities)) {
-            buffer.writeByte(mark);
-            buffer.writeFixString(sqlState);
+//            buffer.writeByte(mark);
+//            buffer.writeFixString(sqlState);
             buffer.writeEOFString(message);
         } else {
             buffer.writeEOFString(message);

config/src/main/java/io/mycat/config/ClusterConfig.java

@@ -24,8 +24,8 @@ public class ClusterConfig {
     private String readBalanceName;
     private String writeBalanceName;
     @javax.validation.constraints.NotNull
-    private List<String> masters;
-    private List<String> replicas;
+    private List<String> masters = new ArrayList<>();
+    private List<String> replicas = new ArrayList<>();
     private HeartbeatConfig heartbeat = HeartbeatConfig.builder()
             .minSwitchTimeInterval(300)
             .heartbeatTimeout(1000)

example/src/test/java/io/mycat/sql/ExceptionCaseTest.java

@@ -0,0 +1,91 @@
+package io.mycat.sql;
+
+import com.alibaba.druid.util.JdbcUtils;
+import io.mycat.assemble.MycatTest;
+import io.mycat.config.DatasourceConfig;
+import io.mycat.config.ShardingBackEndTableInfoConfig;
+import io.mycat.config.ShardingFunction;
+import io.mycat.config.ShardingTableConfig;
+import io.mycat.hint.CreateClusterHint;
+import io.mycat.hint.CreateDataSourceHint;
+import io.mycat.hint.CreateSchemaHint;
+import io.mycat.hint.CreateTableHint;
+import io.mycat.router.mycat1xfunction.PartitionByFileMap;
+import io.mycat.router.mycat1xfunction.PartitionByHotDate;
+import io.mycat.util.ByteUtil;
+import io.vertx.core.json.Json;
+import org.apache.groovy.util.Maps;
+import org.junit.Assert;
+import org.junit.Test;
+
+import javax.annotation.concurrent.NotThreadSafe;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.sql.*;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+
+@NotThreadSafe
+@net.jcip.annotations.NotThreadSafe
+public class ExceptionCaseTest implements MycatTest {
+
+    @Test
+    public void case2() throws Exception {
+        try (Connection mycatConnection = getMySQLConnection(DB_MYCAT)) {
+            execute(mycatConnection, RESET_CONFIG);
+
+            execute(mycatConnection, "DROP DATABASE db1");
+
+
+            execute(mycatConnection, "CREATE DATABASE db1");
+
+
+            execute(mycatConnection, CreateDataSourceHint
+                    .create("ds0",
+                            DB1));
+
+            execute(mycatConnection,
+                    CreateClusterHint.create("c0",
+                            Arrays.asList("ds0"), Collections.emptyList()));
+
+            execute(mycatConnection, "USE `db1`;");
+
+            execute(mycatConnection, "CREATE TABLE if not exists `company` ( `id` int(11) NOT NULL AUTO_INCREMENT,`companyname` varchar(20) DEFAULT NULL,`addressid` int(11) DEFAULT NULL,PRIMARY KEY (`id`))");
+
+            execute(mycatConnection, "delete from db1.company");
+
+            execute(mycatConnection, "INSERT INTO `db1`.`company` (id,`companyname`,`addressid`) VALUES (1,'Intel',1)");
+
+            try {
+                execute(mycatConnection, "INSERT INTO `db1`.`company` (id,`companyname`,`addressid`) VALUES (1,'Intel',1)");
+            } catch (SQLException e) {
+                Assert.assertEquals(1062,e.getErrorCode());
+                Assert.assertEquals("23000",e.getSQLState());
+                Assert.assertEquals("Duplicate entry '1' for key 'PRIMARY'",e.getMessage());
+                System.out.println();
+            }
+
+
+            execute(mycatConnection, "CREATE TABLE if not exists `company` ( `id` int(11) NOT NULL AUTO_INCREMENT,`companyname` varchar(20) DEFAULT NULL,`addressid` int(11) DEFAULT NULL,PRIMARY KEY (`id`))"
+                    + " dbpartition by mod_hash(id) tbpartition by mod_hash(id) tbpartitions 2 dbpartitions 2;");
+
+            execute(mycatConnection, "delete from db1.company");
+
+            execute(mycatConnection, "INSERT INTO `db1`.`company` (id,`companyname`,`addressid`) VALUES (1,'Intel',1)");
+
+            try {
+                execute(mycatConnection, "INSERT INTO `db1`.`company` (id,`companyname`,`addressid`) VALUES (1,'Intel',1)");
+            } catch (SQLException e) {
+                Assert.assertEquals(1062,e.getErrorCode());
+                Assert.assertEquals("23000",e.getSQLState());
+                Assert.assertEquals("Duplicate entry '1' for key 'PRIMARY'",e.getMessage());
+                System.out.println();
+            }
+        }
+    }
+
+}

mycat2/src/main/java/io/mycat/mycatmysql/MycatVertxMySQLHandler.java

@@ -38,6 +38,7 @@
 import java.io.ByteArrayOutputStream;
 import java.nio.charset.StandardCharsets;
 import java.sql.JDBCType;
+import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -338,8 +339,23 @@ public void handle0(int packetId, Buffer event, NetSocket socket, Process proces
             }
             promise.onComplete(o -> {
                 if (o.failed()) {
-                    mycatDataContext.setLastMessage(o.cause());
-                    this.session.writeErrorEndPacketBySyncInProcessError(0);
+                    Throwable cause = o.cause();
+                    int errorCode = 0;
+                    String message;
+                    String sqlState;
+                    if (cause instanceof SQLException){
+                        errorCode = ((SQLException) cause).getErrorCode();
+                        message = ((SQLException) cause).getMessage();
+                        sqlState = ((SQLException) cause).getSQLState();
+                    }else if (cause instanceof MycatException){
+                        errorCode = ((MycatException) cause).getErrorCode();
+                        message = ((MycatException) cause).getMessage();
+                        sqlState = "";
+                    }else {
+                        message = o.toString();
+                    }
+                    mycatDataContext.setLastMessage(message);
+                    this.session.writeErrorEndPacketBySyncInProcessError(errorCode);
                 }
                 checkPendingMessages();
             });