Repository: apache/shardingsphere
Issue: Exception fired during concurrently query

## Bug Report

**For English only**, other languages will not accept.

Before report a bug, make sure you have:

- Searched open and closed [GitHub issues](https://github.com/sharding-sphere/sharding-sphere/issues).
- Read documentation: [ShardingSphere Doc](http://shardingsphere.io/document/current/en/overview/).

Please pay attention on issues you submitted, because we maybe need more details. 
If no response **more than 7 days** and we cannot reproduce it on current information, we will **close it**.

Please answer these questions before submitting your issue. Thanks!

### Which version of ShardingSphere did you use?
4.0.0-M1
### Which project did you use? Sharding-JDBC or Sharding-Proxy?
Sharding-Proxy
### Expected behavior
Execute query correctly.
### Actual behavior

Exception 1:
java.lang.NullPointerException: null
        at com.mysql.jdbc.ResultSetImpl.checkColumnBounds(ResultSetImpl.java:766)
        at com.mysql.jdbc.ResultSetImpl.getObject(ResultSetImpl.java:4420)
        at com.zaxxer.hikari.pool.HikariProxyResultSet.getObject(HikariProxyResultSet.java)
        at org.apache.shardingsphere.core.executor.sql.execute.result.StreamQueryResult.getValue(StreamQueryResult.java:75)
        at org.apache.shardingsphere.core.merger.dql.common.StreamMergedResult.getValue(StreamMergedResult.java:49)
        at org.apache.shardingsphere.shardingproxy.backend.communication.jdbc.JDBCDatabaseCommunicationEngine.getQueryData(JDBCDatabaseCommunicationEngine.java:149)
        at org.apache.shardingsphere.shardingproxy.transport.mysql.packet.command.query.binary.execute.MySQLQueryComStmtExecutePacketExecutor.getQueryData(MySQLQueryComStmtExecutePacketExecutor.java:1
13)
        at org.apache.shardingsphere.shardingproxy.transport.mysql.packet.command.query.binary.execute.MySQLQueryComStmtExecutePacketExecutor.getQueryData(MySQLQueryComStmtExecutePacketExecutor.java:5
3)
        at org.apache.shardingsphere.shardingproxy.frontend.mysql.MySQLFrontendEngine.writeMoreResults(MySQLFrontendEngine.java:152)
        at org.apache.shardingsphere.shardingproxy.frontend.mysql.MySQLFrontendEngine.writePackets(MySQLFrontendEngine.java:133)
        at org.apache.shardingsphere.shardingproxy.frontend.mysql.MySQLFrontendEngine.executeCommand(MySQLFrontendEngine.java:108)
        at org.apache.shardingsphere.shardingproxy.frontend.common.netty.FrontendChannelInboundHandler$1.run(FrontendChannelInboundHandler.java:70)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)










Exception 2:
java.sql.SQLException: Operation not allowed after ResultSet closed
        at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:965)
        at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:898)
        at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:887)
        at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:861)
        at com.mysql.jdbc.ResultSetImpl.checkClosed(ResultSetImpl.java:743)
        at com.mysql.jdbc.ResultSetImpl.next(ResultSetImpl.java:6289)
        at com.zaxxer.hikari.pool.HikariProxyResultSet.next(HikariProxyResultSet.java)
        at org.apache.shardingsphere.core.executor.sql.execute.result.StreamQueryResult.next(StreamQueryResult.java:68)
        at org.apache.shardingsphere.core.merger.dql.iterator.IteratorStreamMergedResult.next(IteratorStreamMergedResult.java:43)
        at org.apache.shardingsphere.shardingproxy.backend.communication.jdbc.JDBCDatabaseCommunicationEngine.next(JDBCDatabaseCommunicationEngine.java:141)
        at org.apache.shardingsphere.shardingproxy.transport.mysql.packet.command.query.binary.execute.MySQLQueryComStmtExecutePacketExecutor.next(MySQLQueryComStmtExecutePacketExecutor.java:108)
        at org.apache.shardingsphere.shardingproxy.frontend.mysql.MySQLFrontendEngine.writeMoreResults(MySQLFrontendEngine.java:147)
        at org.apache.shardingsphere.shardingproxy.frontend.mysql.MySQLFrontendEngine.writePackets(MySQLFrontendEngine.java:134)
        at org.apache.shardingsphere.shardingproxy.frontend.mysql.MySQLFrontendEngine.executeCommand(MySQLFrontendEngine.java:108)
        at org.apache.shardingsphere.shardingproxy.frontend.common.netty.FrontendChannelInboundHandler$1.run(FrontendChannelInboundHandler.java:70)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)












Exception 3:
[ERROR] 18:30:44.867 [ShardingSphere-Command-20] o.a.s.s.f.c.n.FrontendChannelInboundHandler - Exception occur:
java.sql.SQLException: null
        at org.apache.shardingsphere.shardingproxy.backend.communication.jdbc.connection.BackendConnection.throwSQLExceptionIfNecessary(BackendConnection.java:296)
        at org.apache.shardingsphere.shardingproxy.backend.communication.jdbc.connection.BackendConnection.close(BackendConnection.java:246)
        at org.apache.shardingsphere.shardingproxy.backend.communication.jdbc.connection.BackendConnection.close(BackendConnection.java:228)
        at org.apache.shardingsphere.shardingproxy.frontend.common.netty.FrontendChannelInboundHandler$1.run(FrontendChannelInboundHandler.java:73)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

### Reason analyze (If you can)
MySQL connection closed during query. Defect of BackendConnection status management.
### Steps to reproduce the behavior, such as: SQL to execute, sharding rule configuration, when exception occur etc.
200 connections concurrently execute SQLs.
### Example codes for reproduce this issue (such as a github link).


Changed files:
sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/ConnectionStateHandler.java
sharding-proxy/sharding-proxy-frontend/sharding-proxy-frontend-core/src/main/java/org/apache/shardingsphere/shardingproxy/frontend/command/CommandExecutorTask.java
sharding-proxy/sharding-proxy-backend/src/test/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnectionTest.java
sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnection.java

diff --git a/sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnection.java b/sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnection.java
index 36099314411..0a1b6e687e6 100644
--- a/sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnection.java
+++ b/sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnection.java
@@ -132,7 +132,6 @@ public final class BackendConnection implements AutoCloseable {
      * @throws SQLException SQL exception
      */
     public List<Connection> getConnections(final ConnectionMode connectionMode, final String dataSourceName, final int connectionSize) throws SQLException {
-        stateHandler.setRunningStatusIfNecessary();
         if (stateHandler.isInTransaction()) {
             return getConnectionsWithTransaction(connectionMode, dataSourceName, connectionSize);
         } else {
diff --git a/sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/ConnectionStateHandler.java b/sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/ConnectionStateHandler.java
index 18dc8bea4d4..213f785fd30 100644
--- a/sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/ConnectionStateHandler.java
+++ b/sharding-proxy/sharding-proxy-backend/src/main/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/ConnectionStateHandler.java
@@ -57,8 +57,8 @@ public class ConnectionStateHandler {
     /**
      * Change connection status to running if necessary.
      */
-    void setRunningStatusIfNecessary() {
-        if (ConnectionStatus.TRANSACTION != status.get()) {
+    public void setRunningStatusIfNecessary() {
+        if (ConnectionStatus.TRANSACTION != status.get() && ConnectionStatus.RUNNING != status.get()) {
             status.getAndSet(ConnectionStatus.RUNNING);
         }
     }
diff --git a/sharding-proxy/sharding-proxy-backend/src/test/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnectionTest.java b/sharding-proxy/sharding-proxy-backend/src/test/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnectionTest.java
index 3e9ab27416f..04db6ff0a7a 100644
--- a/sharding-proxy/sharding-proxy-backend/src/test/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnectionTest.java
+++ b/sharding-proxy/sharding-proxy-backend/src/test/java/org/apache/shardingsphere/shardingproxy/backend/communication/jdbc/connection/BackendConnectionTest.java
@@ -158,7 +158,8 @@ public final class BackendConnectionTest {
             backendConnection.setCurrentSchema("schema_0");
             when(backendDataSource.getConnections((ConnectionMode) any(), anyString(), eq(12), eq(TransactionType.LOCAL))).thenReturn(MockConnectionUtil.mockNewConnections(12));
             backendConnection.getConnections(ConnectionMode.MEMORY_STRICTLY, "ds1", 12);
-            assertThat(backendConnection.getStateHandler().getStatus(), is(ConnectionStatus.RUNNING));
+            assertThat(backendConnection.getStateHandler().getStatus(), is(ConnectionStatus.INIT));
+            backendConnection.getStateHandler().setRunningStatusIfNecessary();
             mockResultSetAndStatement(backendConnection);
             actual = backendConnection;
         }
diff --git a/sharding-proxy/sharding-proxy-frontend/sharding-proxy-frontend-core/src/main/java/org/apache/shardingsphere/shardingproxy/frontend/command/CommandExecutorTask.java b/sharding-proxy/sharding-proxy-frontend/sharding-proxy-frontend-core/src/main/java/org/apache/shardingsphere/shardingproxy/frontend/command/CommandExecutorTask.java
index 1ecf27029fb..b6117210b1c 100644
--- a/sharding-proxy/sharding-proxy-frontend/sharding-proxy-frontend-core/src/main/java/org/apache/shardingsphere/shardingproxy/frontend/command/CommandExecutorTask.java
+++ b/sharding-proxy/sharding-proxy-frontend/sharding-proxy-frontend-core/src/main/java/org/apache/shardingsphere/shardingproxy/frontend/command/CommandExecutorTask.java
@@ -62,6 +62,7 @@ public final class CommandExecutorTask implements Runnable {
         try (BackendConnection backendConnection = this.backendConnection;
              PacketPayload payload = databaseFrontendEngine.getCodecEngine().createPacketPayload((ByteBuf) message)) {
             backendConnection.getStateHandler().waitUntilConnectionReleasedIfNecessary();
+            backendConnection.getStateHandler().setRunningStatusIfNecessary();
             isNeedFlush = executeCommand(context, payload, backendConnection);
             connectionSize = backendConnection.getConnectionSize();
             // CHECKSTYLE:OFF