From 18b304f99dac5a5feaac52c87e4560f77c552d41 Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 14 Jan 2019 14:25:47 +0800 Subject: [PATCH 1/4] specified mq dynamic topic --- .../adapter/rdb/service/RdbSyncService.java | 18 ++- .../client/adapter/rdb/support/SingleDml.java | 14 +-- .../otter/canal/common/MQMessageUtils.java | 115 ++++++++++++------ .../otter/canal/kafka/CanalKafkaProducer.java | 44 +++---- .../canal/rocketmq/CanalRocketMQProducer.java | 40 +++--- 5 files changed, 127 insertions(+), 104 deletions(-) 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 612ba305..95ac4ef5 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 @@ -189,7 +189,7 @@ public class RdbSyncService { } else if (type != null && type.equalsIgnoreCase("DELETE")) { delete(batchExecutor, config, dml); } else if (type != null && type.equalsIgnoreCase("TRUNCATE")) { - truncate(batchExecutor, config, dml); + truncate(batchExecutor, config); } if (logger.isDebugEnabled()) { logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); @@ -353,15 +353,13 @@ public class RdbSyncService { * * @param config */ - private void truncate(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) throws SQLException { - if (dml.getIsTruncate()) { - DbMapping dbMapping = config.getDbMapping(); - StringBuilder sql = new StringBuilder(); - sql.append("TRUNCATE TABLE ").append(SyncUtil.getDbTableName(dbMapping)); - batchExecutor.execute(sql.toString(), new ArrayList<>()); - if (logger.isTraceEnabled()) { - logger.trace("Truncate target table, sql: {}", sql); - } + private void truncate(BatchExecutor batchExecutor, MappingConfig config) throws SQLException { + DbMapping dbMapping = config.getDbMapping(); + StringBuilder sql = new StringBuilder(); + sql.append("TRUNCATE TABLE ").append(SyncUtil.getDbTableName(dbMapping)); + batchExecutor.execute(sql.toString(), new ArrayList<>()); + if (logger.isTraceEnabled()) { + logger.trace("Truncate target table, sql: {}", sql); } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java index 417f2880..bad12249 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java @@ -14,7 +14,6 @@ public class SingleDml { private String type; private Map data; private Map old; - private boolean isTruncate = false; public String getDestination() { return destination; @@ -64,18 +63,10 @@ public class SingleDml { this.old = old; } - public boolean getIsTruncate() { - return isTruncate; - } - - public void setIsTruncate(boolean isTruncate) { - this.isTruncate = isTruncate; - } - public static List dml2SingleDmls(Dml dml) { - int size = dml.getData() == null ? 0 : dml.getData().size(); List singleDmls = new ArrayList<>(); - if (size > 0) { + if (dml.getData() != null) { + int size = dml.getData().size(); for (int i = 0; i < size; i++) { SingleDml singleDml = new SingleDml(); singleDml.setDestination(dml.getDestination()); @@ -94,7 +85,6 @@ public class SingleDml { singleDml.setDatabase(dml.getDatabase()); singleDml.setTable(dml.getTable()); singleDml.setType(dml.getType()); - singleDml.setIsTruncate(true); singleDmls.add(singleDml); } return singleDmls; diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java index c767f82f..bf4c1921 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -1,12 +1,6 @@ package com.alibaba.otter.canal.common; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; import org.apache.commons.lang.StringUtils; @@ -30,36 +24,46 @@ import com.google.protobuf.InvalidProtocolBufferException; public class MQMessageUtils { @SuppressWarnings("deprecation") - private static Map> partitionDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(), - new Function>() { + private static Map> partitionDatas = MigrateMap + .makeComputingMap(new MapMaker().softValues(), new Function>() { public List apply(String pkHashConfigs) { - List datas = Lists.newArrayList(); - String[] pkHashConfigArray = StringUtils.split(pkHashConfigs, - ","); + List datas = Lists + .newArrayList(); + String[] pkHashConfigArray = StringUtils + .split(pkHashConfigs, ","); // schema.table:id^name for (String pkHashConfig : pkHashConfigArray) { PartitionData data = new PartitionData(); - int i = pkHashConfig.lastIndexOf(":"); + int i = pkHashConfig + .lastIndexOf(":"); if (i > 0) { - String pkStr = pkHashConfig.substring(i + 1); - if (pkStr.equalsIgnoreCase("$pk$")) { + String pkStr = pkHashConfig + .substring(i + 1); + if (pkStr.equalsIgnoreCase( + "$pk$")) { data.hashMode.autoPkHash = true; } else { - data.hashMode.pkNames = Lists.newArrayList(StringUtils.split(pkStr, - '^')); + data.hashMode.pkNames = Lists + .newArrayList( + StringUtils + .split( + pkStr, + '^')); } - pkHashConfig = pkHashConfig.substring(0, - i); + pkHashConfig = pkHashConfig + .substring(0, i); } else { data.hashMode.tableHash = true; } - if (!isWildCard(pkHashConfig)) { + if (!isWildCard( + pkHashConfig)) { data.simpleName = pkHashConfig; } else { - data.regexFilter = new AviaterRegexFilter(pkHashConfig); + data.regexFilter = new AviaterRegexFilter( + pkHashConfig); } datas.add(data); } @@ -69,24 +73,29 @@ public class MQMessageUtils { }); @SuppressWarnings("deprecation") - private static Map> dynamicTopicDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(), - new Function>() { + private static Map> dynamicTopicDatas = MigrateMap + .makeComputingMap(new MapMaker().softValues(), new Function>() { public List apply(String pkHashConfigs) { - List datas = Lists.newArrayList(); - String[] dynamicTopicArray = StringUtils.split(pkHashConfigs, - ","); + List datas = Lists + .newArrayList(); + String[] dynamicTopicArray = StringUtils + .split(pkHashConfigs, ","); // schema.table for (String dynamicTopic : dynamicTopicArray) { DynamicTopicData data = new DynamicTopicData(); - if (!isWildCard(dynamicTopic)) { + if (!isWildCard( + dynamicTopic)) { data.simpleName = dynamicTopic; } else { - if (dynamicTopic.contains("\\.")) { - data.tableRegexFilter = new AviaterRegexFilter(dynamicTopic); + if (dynamicTopic + .contains("\\.")) { + data.tableRegexFilter = new AviaterRegexFilter( + dynamicTopic); } else { - data.schemaRegexFilter = new AviaterRegexFilter(dynamicTopic); + data.schemaRegexFilter = new AviaterRegexFilter( + dynamicTopic); } } datas.add(data); @@ -134,15 +143,22 @@ public class MQMessageUtils { if (StringUtils.isEmpty(schemaName) || StringUtils.isEmpty(tableName)) { put2MapMessage(messages, message.getId(), defaultTopic, entry); } else { - if (matchDynamicTopic(schemaName + "." + tableName, dynamicTopicConfigs)) { - put2MapMessage(messages, message.getId(), schemaName + "." + tableName, entry); - } else if (matchDynamicTopic(schemaName, dynamicTopicConfigs)) { - put2MapMessage(messages, message.getId(), schemaName, entry); + Set topics = matchTopics(schemaName + "." + tableName, dynamicTopicConfigs); + if (topics != null) { + for (String topic : topics) { + put2MapMessage(messages, message.getId(), topic, entry); + } } else { - put2MapMessage(messages, message.getId(), defaultTopic, entry); + topics = matchTopics(schemaName, dynamicTopicConfigs); + if (topics != null) { + for (String topic : topics) { + put2MapMessage(messages, message.getId(), topic, entry); + } + } else { + put2MapMessage(messages, message.getId(), defaultTopic, entry); + } } } - } return messages; } @@ -275,8 +291,9 @@ public class MQMessageUtils { try { rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { - throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" - + entry.toString(), e); + throw new RuntimeException( + "ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), + e); } CanalEntry.EventType eventType = rowChange.getEventType(); @@ -480,6 +497,24 @@ public class MQMessageUtils { return null; } + private static Set matchTopics(String name, String dynamicTopicConfigs) { + String[] router = StringUtils.split(dynamicTopicConfigs, ';'); + Set topics = new HashSet<>(); + for (String item : router) { + int i = item.indexOf(":"); + if (i > -1) { + String topic = item.substring(0, i).trim(); + String topicConfigs = item.substring(i + 1).trim(); + if (matchDynamicTopic(name, topicConfigs)) { + topics.add(topic); + } + } else if (matchDynamicTopic(name, item)) { + topics.add(name); + } + } + return topics.isEmpty() ? null : topics; + } + public static boolean matchDynamicTopic(String name, String dynamicTopicConfigs) { if (StringUtils.isEmpty(dynamicTopicConfigs)) { return false; @@ -520,8 +555,8 @@ public class MQMessageUtils { private static boolean isWildCard(String value) { // not contaiins '.' ? - return StringUtils.containsAny(value, new char[] { '*', '?', '+', '|', '(', ')', '{', '}', '[', ']', '\\', '$', - '^' }); + return StringUtils.containsAny(value, + new char[] { '*', '?', '+', '|', '(', ')', '{', '}', '[', ']', '\\', '$', '^' }); } private static void put2MapMessage(Map messageMap, Long messageId, String topicName, diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 611710d4..b22c852c 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -79,27 +79,31 @@ public class CanalKafkaProducer implements CanalMQProducer { @Override public void send(MQProperties.CanalDestination canalDestination, Message message, Callback callback) { + try { + if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) { + // 动态topic + Map messageMap = MQMessageUtils + .messageTopics(message, canalDestination.getTopic(), canalDestination.getDynamicTopic()); - if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) { - // 动态topic - Map messageMap = MQMessageUtils - .messageTopics(message, canalDestination.getTopic(), canalDestination.getDynamicTopic()); - - for (Map.Entry entry : messageMap.entrySet()) { - String topicName = entry.getKey().replace('.', '_'); - Message messageSub = entry.getValue(); - if (logger.isDebugEnabled()) { - logger.debug("## Send message to kafka topic: " + topicName); + for (Map.Entry entry : messageMap.entrySet()) { + String topicName = entry.getKey().replace('.', '_'); + Message messageSub = entry.getValue(); + if (logger.isDebugEnabled()) { + logger.debug("## Send message to kafka topic: " + topicName); + } + send(canalDestination, topicName, messageSub); } - send(canalDestination, topicName, messageSub, callback); + } else { + send(canalDestination, canalDestination.getTopic(), message); } - } else { - send(canalDestination, canalDestination.getTopic(), message, callback); + callback.commit(); + } catch (Exception e) { + callback.rollback(); } } - private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message, - Callback callback) { + private void send(MQProperties.CanalDestination canalDestination, String topicName, + Message message) throws Exception { // producer.beginTransaction(); if (!kafkaProperties.getFlatMessage()) { try { @@ -134,8 +138,7 @@ public class CanalKafkaProducer implements CanalMQProducer { } catch (Exception e) { logger.error(e.getMessage(), e); // producer.abortTransaction(); - callback.rollback(); - return; + throw e; } } else { // 发送扁平数据json @@ -156,8 +159,7 @@ public class CanalKafkaProducer implements CanalMQProducer { } catch (Exception e) { logger.error(e.getMessage(), e); // producer.abortTransaction(); - callback.rollback(); - return; + throw e; } } else { FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, @@ -177,8 +179,7 @@ public class CanalKafkaProducer implements CanalMQProducer { } catch (Exception e) { logger.error(e.getMessage(), e); // producer.abortTransaction(); - callback.rollback(); - return; + throw e; } } } @@ -193,7 +194,6 @@ public class CanalKafkaProducer implements CanalMQProducer { } // producer.commitTransaction(); - callback.commit(); } } diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index 559704d1..ff7db991 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -61,23 +61,28 @@ public class CanalRocketMQProducer implements CanalMQProducer { @Override public void send(final MQProperties.CanalDestination destination, com.alibaba.otter.canal.protocol.Message data, Callback callback) { - if (!StringUtils.isEmpty(destination.getDynamicTopic())) { - // 动态topic - Map messageMap = MQMessageUtils - .messageTopics(data, destination.getTopic(), destination.getDynamicTopic()); + try { + if (!StringUtils.isEmpty(destination.getDynamicTopic())) { + // 动态topic + Map messageMap = MQMessageUtils + .messageTopics(data, destination.getTopic(), destination.getDynamicTopic()); - for (Map.Entry entry : messageMap.entrySet()) { - String topicName = entry.getKey().replace('.', '_'); - com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue(); - send(destination, topicName, messageSub, callback); + for (Map.Entry entry : messageMap.entrySet()) { + String topicName = entry.getKey().replace('.', '_'); + com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue(); + send(destination, topicName, messageSub); + } + } else { + send(destination, destination.getTopic(), data); } - } else { - send(destination, destination.getTopic(), data, callback); + callback.commit(); + } catch (Exception e) { + callback.rollback(); } } public void send(final MQProperties.CanalDestination destination, String topicName, - com.alibaba.otter.canal.protocol.Message data, Callback callback) { + com.alibaba.otter.canal.protocol.Message data) throws Exception { if (!mqProperties.getFlatMessage()) { try { if (destination.getPartition() != null) { @@ -132,8 +137,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { }, null); } catch (Exception e) { logger.error("send flat message to hashed partition error", e); - callback.rollback(); - return; + throw e; } } } @@ -141,8 +145,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { logger.error("Send message error!", e); - callback.rollback(); - return; + throw e; } } else { List flatMessages = MQMessageUtils.messageConverter(data); @@ -167,8 +170,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { }, null); } catch (Exception e) { logger.error("send flat message to fixed partition error", e); - callback.rollback(); - return; + throw e; } } else { if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { @@ -205,8 +207,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { }, null); } catch (Exception e) { logger.error("send flat message to hashed partition error", e); - callback.rollback(); - return; + throw e; } } } @@ -216,7 +217,6 @@ public class CanalRocketMQProducer implements CanalMQProducer { } } - callback.commit(); if (logger.isDebugEnabled()) { logger.debug("send message to rocket topic: {}", destination.getTopic()); } From ca41be8412becc012c410a588c40df18c119d238 Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 15 Jan 2019 11:38:00 +0800 Subject: [PATCH 2/4] fixed #1420 --- .../otter/canal/kafka/CanalKafkaProducer.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index b22c852c..490d50b5 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -144,7 +144,10 @@ public class CanalKafkaProducer implements CanalMQProducer { // 发送扁平数据json List flatMessages = MQMessageUtils.messageConverter(message); if (flatMessages != null) { + int idx = 0; + int size = flatMessages.size(); for (FlatMessage flatMessage : flatMessages) { + idx++; if (StringUtils.isEmpty(canalDestination.getPartitionHash())) { try { Integer partition = canalDestination.getPartition(); @@ -155,7 +158,11 @@ public class CanalKafkaProducer implements CanalMQProducer { partition, null, JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); - producer2.send(record); + if (idx != size) { + producer2.send(record); + } else { + producer2.send(record).get(); + } } catch (Exception e) { logger.error(e.getMessage(), e); // producer.abortTransaction(); @@ -175,7 +182,11 @@ public class CanalKafkaProducer implements CanalMQProducer { i, null, JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue)); - producer2.send(record); + if (idx != size) { + producer2.send(record); + } else { + producer2.send(record).get(); + } } catch (Exception e) { logger.error(e.getMessage(), e); // producer.abortTransaction(); From 60441d8b31dd9892af191ca4bd726777b031cb55 Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 15 Jan 2019 18:20:05 +0800 Subject: [PATCH 3/4] fixed #1396 --- .../adapter/rdb/service/RdbSyncService.java | 11 ++++-- .../adapter/rdb/support/BatchExecutor.java | 37 ++++++++++++++----- 2 files changed, 35 insertions(+), 13 deletions(-) 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 95ac4ef5..bb804e40 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 @@ -73,10 +73,10 @@ public class RdbSyncService { this.executorThreads = new ExecutorService[this.threads]; for (int i = 0; i < this.threads; i++) { dmlsPartition[i] = new ArrayList<>(); - batchExecutors[i] = new BatchExecutor(dataSource.getConnection()); + batchExecutors[i] = new BatchExecutor(dataSource); executorThreads[i] = Executors.newSingleThreadExecutor(); } - } catch (SQLException e) { + } catch (Exception e) { throw new RuntimeException(e); } } @@ -122,6 +122,12 @@ public class RdbSyncService { } }); } + + for (BatchExecutor batchExecutor : batchExecutors) { + if (batchExecutor != null) { + batchExecutor.close(); + } + } } /** @@ -475,7 +481,6 @@ public class RdbSyncService { public void close() { for (int i = 0; i < threads; i++) { - batchExecutors[i].close(); executorThreads[i].shutdown(); } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/BatchExecutor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/BatchExecutor.java index b106e887..e9cf905f 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/BatchExecutor.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/BatchExecutor.java @@ -9,22 +9,38 @@ import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import javax.sql.DataSource; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; +/** + * sql批量执行器 + * + * @author rewerma 2018-11-7 下午06:45:49 + * @version 1.0.0 + */ public class BatchExecutor implements Closeable { private static final Logger logger = LoggerFactory.getLogger(BatchExecutor.class); + private DataSource dataSource; private Connection conn; private AtomicInteger idx = new AtomicInteger(0); - public BatchExecutor(Connection conn) throws SQLException{ - this.conn = conn; - this.conn.setAutoCommit(false); + public BatchExecutor(DataSource dataSource){ + this.dataSource = dataSource; } public Connection getConn() { + if (conn == null) { + try { + conn = dataSource.getConnection(); + this.conn.setAutoCommit(false); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } + } return conn; } @@ -36,7 +52,7 @@ public class BatchExecutor implements Closeable { } public void execute(String sql, List> values) throws SQLException { - PreparedStatement pstmt = conn.prepareStatement(sql); + PreparedStatement pstmt = getConn().prepareStatement(sql); int len = values.size(); for (int i = 0; i < len; i++) { int type = (Integer) values.get(i).get("type"); @@ -49,7 +65,7 @@ public class BatchExecutor implements Closeable { } public void commit() throws SQLException { - conn.commit(); + getConn().commit(); if (logger.isTraceEnabled()) { logger.trace("Batch executor commit " + idx.get() + " rows"); } @@ -57,7 +73,7 @@ public class BatchExecutor implements Closeable { } public void rollback() throws SQLException { - conn.rollback(); + getConn().rollback(); if (logger.isTraceEnabled()) { logger.trace("Batch executor rollback " + idx.get() + " rows"); } @@ -68,10 +84,11 @@ public class BatchExecutor implements Closeable { public void close() { if (conn != null) { try { - if (conn != null) { - conn.close(); - } - } catch (SQLException ioe) { + conn.close(); + } catch (SQLException e) { + logger.error(e.getMessage(), e); + } finally { + conn = null; } } } From 739f3b4fa1622155bbb55bac7e4f7b1d4dc8df7d Mon Sep 17 00:00:00 2001 From: mcy Date: Tue, 15 Jan 2019 18:39:32 +0800 Subject: [PATCH 4/4] fixed #1433 --- .../canal/client/adapter/rdb/service/RdbSyncService.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) 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 bb804e40..0448bcc5 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 @@ -296,15 +296,16 @@ public class RdbSyncService { StringBuilder updateSql = new StringBuilder(); updateSql.append("UPDATE ").append(SyncUtil.getDbTableName(dbMapping)).append(" SET "); List> values = new ArrayList<>(); + boolean hasMatched = false; for (String srcColumnName : old.keySet()) { List targetColumnNames = new ArrayList<>(); columnsMap.forEach((targetColumn, srcColumn) -> { - if (srcColumnName.toLowerCase().equals(srcColumn)) { + if (srcColumnName.toLowerCase().equals(srcColumn.toLowerCase())) { targetColumnNames.add(targetColumn); } }); if (!targetColumnNames.isEmpty()) { - + hasMatched = true; for (String targetColumnName : targetColumnNames) { updateSql.append(targetColumnName).append("=?, "); Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase()); @@ -315,6 +316,10 @@ public class RdbSyncService { } } } + if (!hasMatched) { + logger.warn("Did not matched any columns to update "); + return; + } int len = updateSql.length(); updateSql.delete(len - 2, len).append(" WHERE ");