inner-1945-supplement: report error packet when connection closed (#3562)

inner-2062: fix npe

Signed-off-by: dcy10000 <[email protected]>

Signed-off-by: dcy10000 <[email protected]>
(cherry picked from commit 74e2023)

Author: dcy10000

src/main/java/com/actiontech/dble/services/BackendService.java

@@ -191,6 +191,18 @@ private void handleTaskQueue(ThreadContext threadContext) {
         }
     }
 
+    void parseErrorPacket(byte[] data, String reason) {
+        try {
+            ErrorPacket errPkg = new ErrorPacket();
+            errPkg.read(data);
+            String errMsg = "errNo:" + errPkg.getErrNo() + " " + new String(errPkg.getMessage());
+            LOGGER.warn("no handler process the execute packet err,sql error:{},back service:{},from reason:{}", errMsg, this, reason);
+
+        } catch (RuntimeException e) {
+            LOGGER.info("error handle error-packet", e);
+        }
+    }
+
     /**
      * handle mysql packet returned from backend mysql
      *
@@ -199,6 +211,9 @@ private void handleTaskQueue(ThreadContext threadContext) {
     @Override
     protected void handleInnerData(byte[] data) {
         if (connection.isClosed()) {
+            if (data != null && data.length > 4 && data[4] == ErrorPacket.FIELD_COUNT) {
+                parseErrorPacket(data, "connection close");
+            }
             return;
         }
 
@@ -218,16 +233,28 @@ protected void handleDataError(Exception e) {
         LOGGER.warn(this.toString() + " handle data error:", e);
         connection.close("handle data error:" + e.getMessage());
         while (taskQueue.size() > 0) {
-            taskQueue.clear();
+            clearTaskQueue();
             readSize.set(0);
             // clear all data from the client
             LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1000));
         }
     }
 
+    private void clearTaskQueue() {
+        ServiceTask task;
+        while ((task = taskQueue.poll()) != null) {
+            if (task.getType() == ServiceTaskType.NORMAL) {
+                final byte[] data = ((NormalServiceTask) task).getOrgData();
+                if (data != null && data.length > 4 && data[4] == ErrorPacket.FIELD_COUNT) {
+                    parseErrorPacket(data, "cleanup");
+                }
+            }
+        }
+    }
+
     @Override
     public void cleanup() {
-        this.taskQueue.clear();
+        clearTaskQueue();
         readSize.set(0);
         backendSpecialCleanUp();
         TraceManager.sessionFinish(this);