check value when update checkpoint (#49)

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

@@ -342,6 +342,8 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
                     break;
                 } catch (InterruptedException e) {
                     // do nothing
+                } catch (IllegalArgumentException e) {
+                    logger.error("Failed to update checkpoint for log message: " + logMessage, e);
                 }
             }
 

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

@@ -296,8 +296,15 @@ public class ClientStream {
      * @param checkpointString Checkpoint string.
      */
     public void setCheckpointString(String checkpointString) {
+        long timestamp = Long.parseLong(checkpointString);
+        if (timestamp < 0) {
+            throw new IllegalArgumentException(
+                    "Update checkpoint with invalid value: " + timestamp);
+        }
+        if (this.checkpointString == null || Long.parseLong(this.checkpointString) < timestamp) {
             this.checkpointString = checkpointString;
         }
+    }
 
     /**
      * Add a {@link RecordListener} to {@link #listeners}.