change data parsing errors to retryable errors (#64)

* data parsing errors changed to retryable errors

* The client can continue to retry unless it exceeds the number of retries or actively exits
Co-authored-by: N花轻 <shimengfei.smf@oceanbase.com>

oblogclient-logproxy/src/main/java/com/oceanbase/clogproxy/client/connection/ClientHandler.java

@@ -225,7 +225,8 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
             logger.error("LogProxy refused handshake request: {}", response.toString());
             throw new LogProxyClientException(
                     ErrorCode.NO_AUTH,
-                    "LogProxy refused handshake request: " + response.toString());
+                    "LogProxy refused handshake request: " + response.toString(),
+                    true);
         } else {
             dataNotEnough = true;
         }

oblogclient-logproxy/src/main/java/com/oceanbase/clogproxy/client/connection/ClientHandlerV01.java

@@ -140,7 +140,9 @@ public class ClientHandlerV01 {
                     resetState();
                     logger.error("LogProxy refused handshake request: {}", message);
                     throw new LogProxyClientException(
-                            ErrorCode.NO_AUTH, "LogProxy refused handshake request: " + message);
+                            ErrorCode.NO_AUTH,
+                            "LogProxy refused handshake request: " + message,
+                            true);
                 } else {
                     dataNotEnough = true;
                 }

oblogclient-logproxy/src/main/java/com/oceanbase/clogproxy/client/connection/ClientStream.java

@@ -144,7 +144,8 @@ public class ClientStream {
                                         triggerException(
                                                 new LogProxyClientException(
                                                         ErrorCode.E_MAX_RECONNECT,
-                                                        "Exceed max retry times"));
+                                                        "Exceed max retry times",
+                                                        true));
                                         break;
                                     }
                                     if (state == ReconnectState.RETRY) {

oblogclient-logproxy/src/main/java/com/oceanbase/clogproxy/client/exception/LogProxyClientException.java

@@ -10,7 +10,6 @@ See the Mulan PSL v2 for more details. */
 
 package com.oceanbase.clogproxy.client.exception;
 
-
 import com.oceanbase.clogproxy.client.enums.ErrorCode;
 
 /**
@@ -22,6 +21,9 @@ public class LogProxyClientException extends RuntimeException {
     /** Error code. */
     private ErrorCode code = ErrorCode.NONE;
 
+    /*The flag of whether the client should stop the stream.*/
+    private boolean needStop = false;
+
     /**
      * Constructor with error code and message.
      *
@@ -56,19 +58,54 @@ public class LogProxyClientException extends RuntimeException {
         this.code = code;
     }
 
+    /**
+     * Constructor with error code and message.
+     *
+     * @param code Error code.
+     * @param message Error message.
+     * @param needStop The flag of whether the client should stop the stream.
+     */
+    public LogProxyClientException(ErrorCode code, String message, boolean needStop) {
+        super(message);
+        this.code = code;
+        this.needStop = needStop;
+    }
+
+    /**
+     * Constructor with error code and exception.
+     *
+     * @param code Error code.
+     * @param exception Exception instance.
+     * @param needStop The flag of whether the client should stop the stream.
+     */
+    public LogProxyClientException(ErrorCode code, Exception exception, boolean needStop) {
+        super(exception.getMessage(), exception.getCause());
+        this.code = code;
+        this.needStop = needStop;
+    }
+
+    /**
+     * Constructor with error code, error message and cause.
+     *
+     * @param code Error code.
+     * @param message Error message.
+     * @param throwable Cause.
+     * @param needStop The flag of whether the client should stop the stream.
+     */
+    public LogProxyClientException(
+        ErrorCode code, String message, Throwable throwable, boolean needStop) {
+        super(message, throwable);
+        this.code = code;
+        this.needStop = needStop;
+    }
+
     /**
      * Identify whether the client should stop the stream.
      *
      * @return The flag of whether the client should stop the stream.
      */
     public boolean needStop() {
-        return (code == ErrorCode.E_MAX_RECONNECT)
-                || (code == ErrorCode.E_PROTOCOL)
-                || (code == ErrorCode.E_HEADER_TYPE)
-                || (code == ErrorCode.NO_AUTH)
-                || (code == ErrorCode.E_COMPRESS_TYPE)
-                || (code == ErrorCode.E_LEN)
-                || (code == ErrorCode.E_PARSE);
+        return needStop;
     }
 
     /**