fixed issue #330 & #350 , json length & escapse

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java

@@ -363,12 +363,12 @@ public class JsonConversion {
                         buf.append('"').append(text).append('"');
                     } else {
                         text = m_data.getFixString((int) m_length, charsetName);
-                        buf.append('"').append(text).append('"');
+                        buf.append('"').append(escapse(text)).append('"');
                     }
 
                     break;
                 case STRING:
-                    buf.append('"').append(m_string_value).append('"');
+                    buf.append('"').append(escapse(m_string_value)).append('"');
                     break;
                 case ERROR:
                     throw new IllegalArgumentException("illegal json data");
@@ -378,6 +378,21 @@ public class JsonConversion {
         }
     }
 
+    private static StringBuilder escapse(String data) {
+        StringBuilder sb = new StringBuilder(data.length());
+        int endIndex = data.length();
+        for (int i = 0; i < endIndex; ++i) {
+            char c = data.charAt(i);
+            if (c == '"') {
+                sb.append('\\');
+            } else if (c == '\\') {
+                sb.append("\\");
+            }
+            sb.append(c);
+        }
+        return sb;
+    }
+
     public static enum Json_enum_type {
         OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE, LITERAL_FALSE, OPAQUE, ERROR
     }

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java

@@ -949,8 +949,28 @@ public final class RowsLogBuffer {
                 break;
             }
             case LogEvent.MYSQL_TYPE_JSON: {
-                len = buffer.getUint16();
-                buffer.forward(meta - 2);
+                switch (meta) {
+                    case 1: {
+                        len = buffer.getUint8();
+                        break;
+                    }
+                    case 2: {
+                        len = buffer.getUint16();
+                        break;
+                    }
+                    case 3: {
+                        len = buffer.getUint24();
+                        break;
+                    }
+                    case 4: {
+                        len = (int) buffer.getUint32();
+                        break;
+                    }
+                    default:
+                        throw new IllegalArgumentException("!! Unknown JSON packlen = " + meta);
+                }
+                // len = buffer.getUint16();
+                // buffer.forward(meta - 4);
                 int position = buffer.position();
                 Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1, charsetName);
                 StringBuilder builder = new StringBuilder();

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java

@@ -275,7 +275,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
         parseThread.setUncaughtExceptionHandler(handler);
         parseThread.setName(String.format("destination = %s , address = %s , EventParser",
             destination,
-            runningInfo == null ? null : runningInfo.getAddress().toString()));
+            runningInfo == null ? null : runningInfo.getAddress()));
         parseThread.start();
     }