[hotfix][mysql] Fix number overflow in ObjectUtils.minus (#598)

docs/docs_site.sh

@@ -33,8 +33,10 @@ function start_docs_server() {
 }
 
 function stop_docs_server() {
+  project_dir="$(dirname "$(pwd)")"
   echo "stopping docs server....."
   docker stop ${docs_container_name}
+  rm -rf ${project_dir}/docs/_build
   echo "stop docs server successfully."
 }
 

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/utils/ObjectUtils.java

@@ -24,7 +24,10 @@ import java.math.BigInteger;
 /** Utilities for operation on {@link Object}. */
 public class ObjectUtils {
 
-    /** Returns a number {@code Object} whose value is {@code (number + augend)}. */
+    /**
+     * Returns a number {@code Object} whose value is {@code (number + augend)}, Note: This method
+     * does not consider number overflow because we don't want to change the object type.
+     */
     public static Object plus(Object number, int augend) {
         if (number instanceof Integer) {
             return (int) number + augend;
@@ -52,9 +55,10 @@ public class ObjectUtils {
                             subtrahend.getClass().getSimpleName()));
         }
         if (minuend instanceof Integer) {
-            return BigDecimal.valueOf((int) minuend - (int) subtrahend);
+            return BigDecimal.valueOf((int) minuend).subtract(BigDecimal.valueOf((int) subtrahend));
         } else if (minuend instanceof Long) {
-            return BigDecimal.valueOf((long) minuend - (long) subtrahend);
+            return BigDecimal.valueOf((long) minuend)
+                    .subtract(BigDecimal.valueOf((long) subtrahend));
         } else if (minuend instanceof BigInteger) {
             return new BigDecimal(
                     ((BigInteger) minuend).subtract((BigInteger) subtrahend).toString());

flink-connector-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/source/utils/ObjectUtilsTest.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ververica.cdc.connectors.mysql.source.utils;
+
+import org.junit.Test;
+
+import java.math.BigDecimal;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests for {@link ObjectUtils}. */
+public class ObjectUtilsTest {
+
+    @Test
+    public void testMinus() {
+        assertEquals(BigDecimal.valueOf(9999), ObjectUtils.minus(10000, 1));
+        assertEquals(
+                BigDecimal.valueOf(4294967295L),
+                ObjectUtils.minus(Integer.MAX_VALUE, Integer.MIN_VALUE));
+
+        assertEquals(BigDecimal.valueOf(9999999999999L), ObjectUtils.minus(10000000000000L, 1L));
+        assertEquals(
+                new BigDecimal("18446744073709551615"),
+                ObjectUtils.minus(Long.MAX_VALUE, Long.MIN_VALUE));
+
+        assertEquals(
+                new BigDecimal("99.12344"),
+                ObjectUtils.minus(new BigDecimal("100.12345"), new BigDecimal("1.00001")));
+    }
+}