Merge pull request #1597 from KangZhiDong/master

fix the error log does not contain table name

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java

@@ -76,7 +76,7 @@ public class JdbcTypeUtil {
                || "TEXT".equalsIgnoreCase(columnType) || "TINYTEXT".equalsIgnoreCase(columnType);
     }
 
-    public static Object typeConvert(String columnName, String value, int sqlType, String mysqlType) {
+    public static Object typeConvert(String tableName ,String columnName, String value, int sqlType, String mysqlType) {
         if (value == null
             || (value.equals("") && !(isText(mysqlType) || sqlType == Types.CHAR || sqlType == Types.VARCHAR || sqlType == Types.LONGVARCHAR))) {
             return null;
@@ -161,7 +161,7 @@ public class JdbcTypeUtil {
             }
             return res;
         } catch (Exception e) {
-            logger.error("table: {} column: {}, failed convert type {} to {}", columnName, value, sqlType);
+            logger.error("table: {} column: {}, failed convert type {} to {}", tableName, columnName, value, sqlType);
             return value;
         }
     }

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java

@@ -77,7 +77,7 @@ public class MessageUtil {
                             }
                         }
                         row.put(column.getName(),
-                            JdbcTypeUtil.typeConvert(column.getName(),
+                            JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(),
                                 column.getValue(),
                                 column.getSqlType(),
                                 column.getMysqlType()));
@@ -95,7 +95,7 @@ public class MessageUtil {
                         for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                             if (updateSet.contains(column.getName())) {
                                 rowOld.put(column.getName(),
-                                    JdbcTypeUtil.typeConvert(column.getName(),
+                                    JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(),
                                         column.getValue(),
                                         column.getSqlType(),
                                         column.getMysqlType()));
@@ -153,16 +153,16 @@ public class MessageUtil {
         // }
         List<Map<String, String>> data = flatMessage.getData();
         if (data != null) {
-            dml.setData(changeRows(data, flatMessage.getSqlType(), flatMessage.getMysqlType()));
+            dml.setData(changeRows(dml.getTable(), data, flatMessage.getSqlType(), flatMessage.getMysqlType()));
         }
         List<Map<String, String>> old = flatMessage.getOld();
         if (old != null) {
-            dml.setOld(changeRows(old, flatMessage.getSqlType(), flatMessage.getMysqlType()));
+            dml.setOld(changeRows(dml.getTable(), old, flatMessage.getSqlType(), flatMessage.getMysqlType()));
         }
         return dml;
     }
 
-    private static List<Map<String, Object>> changeRows(List<Map<String, String>> rows, Map<String, Integer> sqlTypes,
+    private static List<Map<String, Object>> changeRows(String table, List<Map<String, String>> rows, Map<String, Integer> sqlTypes,
                                                         Map<String, String> mysqlTypes) {
         List<Map<String, Object>> result = new ArrayList<>();
         for (Map<String, String> row : rows) {
@@ -181,7 +181,7 @@ public class MessageUtil {
                     continue;
                 }
 
-                Object finalValue = JdbcTypeUtil.typeConvert(columnName, columnValue, sqlType, mysqlType);
+                Object finalValue = JdbcTypeUtil.typeConvert(table, columnName, columnValue, sqlType, mysqlType);
                 resultRow.put(columnName, finalValue);
             }
             result.add(resultRow);