Merge pull request #1299 from rewerma/master

FlatMessage没有默认构造方法json反序列化的问题

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

@@ -18,6 +18,7 @@ public class Dml implements Serializable {
     private String                    database;                               // 数据库或schema
     private String                    table;                                  // 表名
     private List<String>              pkNames;
+    private Boolean                   isDdl;
     private String                    type;                                   // 类型: INSERT UPDATE DELETE
     // binlog executeTime
     private Long                      es;                                     // 执行耗时
@@ -59,6 +60,14 @@ public class Dml implements Serializable {
         this.pkNames = pkNames;
     }
 
+    public Boolean getIsDdl() {
+        return isDdl;
+    }
+
+    public void setIsDdl(Boolean isDdl) {
+        this.isDdl = isDdl;
+    }
+
     public String getType() {
         return type;
     }

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

@@ -139,13 +139,14 @@ public class MessageUtil {
         dml.setDatabase(flatMessage.getDatabase());
         dml.setTable(flatMessage.getTable());
         dml.setPkNames(flatMessage.getPkNames());
+        dml.setIsDdl(flatMessage.getIsDdl());
         dml.setType(flatMessage.getType());
         dml.setTs(flatMessage.getTs());
         dml.setEs(flatMessage.getEs());
         dml.setSql(flatMessage.getSql());
-        if (flatMessage.getSqlType() == null || flatMessage.getMysqlType() == null) {
+        // if (flatMessage.getSqlType() == null || flatMessage.getMysqlType() == null) {
-            throw new RuntimeException("SqlType or mysqlType is null");
+        // throw new RuntimeException("SqlType or mysqlType is null");
-        }
+        // }
         List<Map<String, String>> data = flatMessage.getData();
         if (data != null) {
             dml.setData(changeRows(data, flatMessage.getSqlType(), flatMessage.getMysqlType()));

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java

@@ -75,6 +75,11 @@ public class RdbAdapter implements OuterAdapter {
                 rdbMapping.put(key, mappingConfig);
             }
         });
+
+        if (rdbMapping.isEmpty()) {
+            throw new RuntimeException("No rdb adapter found for config key: " + configuration.getKey());
+        }
+
         for (Map.Entry<String, MappingConfig> entry : rdbMapping.entrySet()) {
             String configName = entry.getKey();
             MappingConfig mappingConfig = entry.getValue();
@@ -89,7 +94,7 @@ public class RdbAdapter implements OuterAdapter {
 
                 mirrorDbConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
                                         + mappingConfig.getDbMapping().getDatabase(),
-                        MirrorDbConfig.create(configName, mappingConfig));
+                    MirrorDbConfig.create(configName, mappingConfig));
             }
         }
 

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java

@@ -56,7 +56,7 @@ public class RdbMirrorDbSyncService {
                 if (mirrorDbConfig == null) {
                     continue;
                 }
-                if (StringUtils.isNotEmpty(dml.getSql())) {
+                if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
                     // DDL
                     if (logger.isDebugEnabled()) {
                         logger.debug("DDL: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));

client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java

@@ -135,7 +135,7 @@ public class RdbSyncService {
     public void sync(Map<String, Map<String, MappingConfig>> mappingConfig, List<Dml> dmls) {
         try {
             sync(dmls, dml -> {
-                if (StringUtils.isNotEmpty(dml.getSql())) {
+                if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
                     // DDL
                     columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
                     return false;

protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java

@@ -29,6 +29,9 @@ public class FlatMessage implements Serializable {
     private List<Map<String, String>> data;
     private List<Map<String, String>> old;
 
+    public FlatMessage() {
+    }
+
     public FlatMessage(long id){
         this.id = id;
     }