diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java index cc19813f..38fd7a3b 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java +++ b/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 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; } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java index 87fc2fb9..1ddbae0f 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java +++ b/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) { - throw new RuntimeException("SqlType or mysqlType is null"); - } + // if (flatMessage.getSqlType() == null || flatMessage.getMysqlType() == null) { + // throw new RuntimeException("SqlType or mysqlType is null"); + // } List> data = flatMessage.getData(); if (data != null) { dml.setData(changeRows(data, flatMessage.getSqlType(), flatMessage.getMysqlType())); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java index 3c4f8387..79f10955 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/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 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)); } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java index 4a18ce99..cd3c6567 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java +++ b/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)); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 93797093..dd3a901a 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/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> mappingConfig, List 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; diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index 50e1cb3f..2b958b1c 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -29,6 +29,9 @@ public class FlatMessage implements Serializable { private List> data; private List> old; + public FlatMessage() { + } + public FlatMessage(long id){ this.id = id; }