tttt

eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/sink/handle/DefaultSinkRecordHandler.java

@@ -105,7 +105,7 @@ private void insert(String sql, Schema schema, DataChanges dataChanges) throws S
             Map<String, String> dataChangesAfter = (Map<String, String>) dataChanges.getAfter();
             schema.getFields().forEach(field -> {
                 Type type = eventMeshDialect.getType(field.getColumn());
-                query.setParameter(index.getAndIncrement(), type.convert4Database(dataChangesAfter.get(field.getName())));
+                query.setParameter(index.getAndIncrement(), type.convert2DatabaseType(dataChangesAfter.get(field.getName())));
             });
 
             final int result = query.executeUpdate();

eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/type/Type.java

@@ -26,19 +26,57 @@
 
 public interface Type {
 
-    void configure(DatabaseDialect<?> eventmeshDialect,Dialect hibernateDialect);
+    /**
+     * Configures the database dialects for the EventMesh and Hibernate.
+     *
+     * @param eventmeshDialect The database dialect for the EventMesh.
+     * @param hibernateDialect The database dialect for Hibernate.
+     */
+    void configure(DatabaseDialect<?> eventmeshDialect, Dialect hibernateDialect);
 
+    /**
+     * Retrieves a list of registration keys.
+     *
+     * @return A list of registration keys as strings.
+     */
     List<String> ofRegistrationKeys();
 
+    /**
+     * Retrieves the default value for a given database dialect and column.
+     *
+     * @param databaseDialect The specific database dialect.
+     * @param column          The column for which to retrieve the default value.
+     * @return The default value for the specified database dialect and column.
+     */
     String getDefaultValue(DatabaseDialect<?> databaseDialect, Column<?> column);
 
+    /**
+     * Returns the type name of the specified column.
+     *
+     * @param column the column object for which to retrieve the type name
+     * @return the type name of the column
+     */
     String getTypeName(Column<?> column);
 
+
+    /**
+     * Returns the query binding with value for the specified column and database dialect.
+     *
+     * @param databaseDialect the database dialect object to determine the query binding
+     * @param column          the column object for which to retrieve the query binding with value
+     * @return the query binding with value for the column and database dialect
+     */
     default String getQueryBindingWithValue(DatabaseDialect<?> databaseDialect, Column<?> column) {
         return databaseDialect.getQueryBindingWithValueCast(column);
     }
 
-    default Object convert4Database(Object value){
+    /**
+     * Converts the given value to the appropriate database type.
+     *
+     * @param value the value to be converted
+     * @return the converted value in the database type
+     */
+    default Object convert2DatabaseType(Object value) {
         return value;
     }
 }

eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/type/eventmesh/BytesEventMeshDataType.java

@@ -21,6 +21,7 @@
 import org.apache.eventmesh.connector.jdbc.type.AbstractType;
 
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.List;
 
 public class BytesEventMeshDataType extends AbstractType<byte[]> {
@@ -31,13 +32,15 @@ private BytesEventMeshDataType() {
         super(byte[].class, SQLType.BINARY, "BYTES");
     }
 
-    /**
-     * @return
-     */
     @Override
     public List<String> ofRegistrationKeys() {
         return Arrays.asList("bytes", getName());
     }
 
-
+    @Override
+    public Object convert2DatabaseType(Object value) {
+        //Jackson default serialize byte[] as base64
+        String strValue = (String) value;
+        return Base64.getDecoder().decode(strValue);
+    }
 }

eventmesh-connectors/eventmesh-connector-jdbc/src/main/java/org/apache/eventmesh/connector/jdbc/utils/JdbcStringUtils.java

@@ -22,20 +22,21 @@
 @UtilityClass
 public class JdbcStringUtils {
 
+    /**
+     * Checks whether the given string is wrapped with a specific set of characters.
+     *
+     * @param possiblyWrapped the string to check
+     * @return true if the string is wrapped with characters '`', "'", or "\""; false otherwise
+     */
     public static boolean isWrapped(String possiblyWrapped) {
-        if (possiblyWrapped.length() < 2) {
+        if (possiblyWrapped == null || possiblyWrapped.length() < 2) {
             return false;
         }
-        if (possiblyWrapped.startsWith("`") && possiblyWrapped.endsWith("`")) {
-            return true;
-        }
-        if (possiblyWrapped.startsWith("'") && possiblyWrapped.endsWith("'")) {
-            return true;
-        }
-        if (possiblyWrapped.startsWith("\"") && possiblyWrapped.endsWith("\"")) {
-            return true;
-        }
-        return false;
+        char firstChar = possiblyWrapped.charAt(0);
+        char lastChar = possiblyWrapped.charAt(possiblyWrapped.length() - 1);
+        return (firstChar == '`' && lastChar == '`')
+            || (firstChar == '\'' && lastChar == '\'')
+            || (firstChar == '\"' && lastChar == '\"');
     }
 
     public static boolean isWrapped(char c) {
@@ -47,25 +48,29 @@ public static String withoutWrapper(String possiblyWrapped) {
     }
 
     /**
-     * Compares two version numbers and returns 1 if the first version number is greater than the second version number, -1 if the first version
-     * number is less than the second version number, and 0 if they are equal.
+     * Compares two version numbers and returns the result as an integer.
      *
-     * @param versionX the first version number to compare
-     * @param versionY the second version number to compare
-     * @return 1 if the first version number is greater than the second version number, -1 if the first version number is less than the second version
-     * number, and 0 if they are equal
+     * @param versionX The first version number to compare.
+     * @param versionY The second version number to compare.
+     * @return An integer value representing the comparison result: -1 if versionX is less than versionY, 0 if versionX is equal to versionY, 1 if
+     * versionX is greater than versionY.
      */
     public static int compareVersion(String versionX, String versionY) {
-        String[] v1 = versionX.split("\\.");
-        String[] v2 = versionY.split("\\.");
-        int length = Math.max(v1.length, v2.length);
-        for (int i = 0; i < length; i++) {
-            int num1 = i < v1.length ? Integer.parseInt(v1[i]) : 0;
-            int num2 = i < v2.length ? Integer.parseInt(v2[i]) : 0;
-            if (num1 != num2) {
-                return num1 > num2 ? 1 : -1;
+        String[] firstVersionParts = versionX.split("\\.");
+        String[] secondVersionParts = versionY.split("\\.");
+        int maxLength = Math.max(firstVersionParts.length, secondVersionParts.length);
+        for (int i = 0; i < maxLength; i++) {
+            int firstVersionNumber = getPartAsNumber(firstVersionParts, i);
+            int secondVersionNumber = getPartAsNumber(secondVersionParts, i);
+            if (firstVersionNumber != secondVersionNumber) {
+                return Integer.signum(firstVersionNumber - secondVersionNumber);
             }
         }
         return 0;
     }
+
+
+    private static int getPartAsNumber(String[] parts, int index) {
+        return index < parts.length ? Integer.parseInt(parts[index]) : 0;
+    }
 }

eventmesh-connectors/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/utils/JdbcStringUtilsTest.java

@@ -52,4 +52,44 @@ public void testIsWrappedWithChar() {
         assertTrue(JdbcStringUtils.isWrapped('\"'));
         assertFalse(JdbcStringUtils.isWrapped('A'));
     }
+
+    @Test
+    public void testCompareVersion() {
+        // Test case 1: versionX is less than versionY
+        String versionX1 = "1.0.0";
+        String versionY1 = "1.1.0";
+        int expected1 = -1;
+        int result1 = JdbcStringUtils.compareVersion(versionX1, versionY1);
+        assertEquals(expected1, result1);
+
+        // Test case 2: versionX is equal to versionY
+        String versionX2 = "1.2.3";
+        String versionY2 = "1.2.3";
+        int expected2 = 0;
+        int result2 = JdbcStringUtils.compareVersion(versionX2, versionY2);
+        assertEquals(expected2, result2);
+
+        // Test case 3: versionX is greater than versionY
+        String versionX3 = "2.0.0";
+        String versionY3 = "1.2.3";
+        int expected3 = 1;
+        int result3 = JdbcStringUtils.compareVersion(versionX3, versionY3);
+        assertEquals(expected3, result3);
+
+        assertEquals(0, JdbcStringUtils.compareVersion("1.0", "1.0"));
+        assertEquals(1, JdbcStringUtils.compareVersion("1.1", "1.0"));
+        assertEquals(-1, JdbcStringUtils.compareVersion("1.0", "1.1"));
+        assertEquals(1, JdbcStringUtils.compareVersion("1.10", "1.9"));
+        assertEquals(-1, JdbcStringUtils.compareVersion("1.9", "1.10"));
+        assertEquals(0, JdbcStringUtils.compareVersion("1.0.0", "1"));
+        assertEquals(0, JdbcStringUtils.compareVersion("1.0.0.0", "1.0"));
+        assertEquals(0, JdbcStringUtils.compareVersion("1.0.0.0", "1"));
+        assertEquals(-1, JdbcStringUtils.compareVersion("1.0.0.0", "1.1"));
+        assertEquals(1, JdbcStringUtils.compareVersion("1.1", "1.0.0.0"));
+        try {
+            assertEquals(0, JdbcStringUtils.compareVersion("1.1", "1.a"));
+        } catch (Exception e) {
+            assertTrue(e instanceof NumberFormatException);
+        }
+    }
 }