Repository: apache/shardingsphere
Issue: Proxy blocked when client connect and close repeatly

For English only, other languages we will close it directly.

Please answer these questions before submitting your issue. Thanks!

### Which version of Sharding-Sphere do you using?
3.0.0.M3

### Which project do you using? Sharding-JDBC or Sharding-Proxy?
Sharding-Proxy

### Expected behavior
High Availability.

### Actual behavior
Proxy blocked when client connect and close repeatly.

### Reason analyze
Only when a client close the connection meanwhile a query is not finish in this connection will triger this bug.
When a frontend connection is closed the connection will become not writable, this cause MySQLFrontendHandler wait and never jump out the loop.
 
### Steps to reproduce the behavior, such as: SQL to execute, sharding rule configuration, when exception occur etc
Execute a slow SQL then close the connection until the connections is used up.

### For bug report, please *MUST* provide the reproduce example codes (such as a github link).
Official example.


Changed files:
sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/mysql/MySQLFrontendHandler.java
sharding-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/connection/BackendConnection.java
sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/common/FrontendHandler.java

diff --git a/sharding-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/connection/BackendConnection.java b/sharding-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/connection/BackendConnection.java
index e5a757f3b04..a3c671d8d01 100644
--- a/sharding-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/connection/BackendConnection.java
+++ b/sharding-proxy/src/main/java/io/shardingsphere/proxy/backend/jdbc/connection/BackendConnection.java
@@ -75,6 +75,18 @@ public final class BackendConnection implements AutoCloseable {
         cachedResultSets.add(resultSet);
     }
     
+    /**
+     * Cancel statement.
+     */
+    public void cancel() {
+        for (Statement each : cachedStatements) {
+            try {
+                each.cancel();
+            } catch (final SQLException ignore) {
+            }
+        }
+    }
+    
     @Override
     public void close() throws SQLException {
         Collection<SQLException> exceptions = new LinkedList<>();
diff --git a/sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/common/FrontendHandler.java b/sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/common/FrontendHandler.java
index bcf08e10851..d645a1fc986 100644
--- a/sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/common/FrontendHandler.java
+++ b/sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/common/FrontendHandler.java
@@ -20,7 +20,9 @@ package io.shardingsphere.proxy.frontend.common;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.shardingsphere.proxy.backend.jdbc.connection.BackendConnection;
 import io.shardingsphere.proxy.frontend.common.executor.ChannelThreadExecutorGroup;
+import lombok.Setter;
 
 /**
  * Frontend handler.
@@ -31,6 +33,9 @@ public abstract class FrontendHandler extends ChannelInboundHandlerAdapter {
     
     private boolean authorized;
     
+    @Setter
+    private BackendConnection backendConnection;
+    
     @Override
     public final void channelActive(final ChannelHandlerContext context) {
         ChannelThreadExecutorGroup.getInstance().register(context.channel().id());
@@ -56,6 +61,7 @@ public abstract class FrontendHandler extends ChannelInboundHandlerAdapter {
     @Override
     public final void channelInactive(final ChannelHandlerContext context) {
         context.fireChannelInactive();
+        backendConnection.cancel();
         ChannelThreadExecutorGroup.getInstance().unregister(context.channel().id());
     }
 }
diff --git a/sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/mysql/MySQLFrontendHandler.java b/sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/mysql/MySQLFrontendHandler.java
index 4d6a2869087..e73dcada467 100644
--- a/sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/mysql/MySQLFrontendHandler.java
+++ b/sharding-proxy/src/main/java/io/shardingsphere/proxy/frontend/mysql/MySQLFrontendHandler.java
@@ -105,6 +105,7 @@ public final class MySQLFrontendHandler extends FrontendHandler {
         public void run() {
             try (MySQLPacketPayload payload = new MySQLPacketPayload(message);
                  BackendConnection backendConnection = new BackendConnection()) {
+                setBackendConnection(backendConnection);
                 CommandPacket commandPacket = getCommandPacket(payload, backendConnection);
                 Optional<CommandResponsePackets> responsePackets = commandPacket.execute();
                 if (!responsePackets.isPresent()) {
@@ -132,9 +133,12 @@ public final class MySQLFrontendHandler extends FrontendHandler {
         }
         
         private void writeMoreResults(final QueryCommandPacket queryCommandPacket, final int headPacketsCount) throws SQLException {
+            if (!context.channel().isActive()) {
+                return;
+            }
             currentSequenceId = headPacketsCount;
             while (queryCommandPacket.next()) {
-                while (!context.channel().isWritable()) {
+                while (!context.channel().isWritable() && context.channel().isActive()) {
                     synchronized (MySQLFrontendHandler.this) {
                         try {
                             MySQLFrontendHandler.this.wait();