diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 9bb56792..dd82cafc 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -75,8 +75,8 @@ public class ESAdapter implements OuterAdapter { // 过滤不匹配的key的配置 esSyncConfigTmp.forEach((key, config) -> { if ((config.getOuterAdapterKey() == null && configuration.getKey() == null) - || (config.getOuterAdapterKey() != null && config.getOuterAdapterKey() - .equalsIgnoreCase(configuration.getKey()))) { + || (config.getOuterAdapterKey() != null + && config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { esSyncConfig.put(key, config); } }); @@ -98,15 +98,11 @@ public class ESAdapter implements OuterAdapter { } String schema = matcher.group(2); - schemaItem.getAliasTableItems() - .values() - .forEach(tableItem -> { - Map esSyncConfigMap = dbTableEsSyncConfig.computeIfAbsent(schema - + "-" - + tableItem.getTableName(), - k -> new HashMap<>()); - esSyncConfigMap.put(configName, config); - }); + schemaItem.getAliasTableItems().values().forEach(tableItem -> { + Map esSyncConfigMap = dbTableEsSyncConfig + .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new HashMap<>()); + esSyncConfigMap.put(configName, config); + }); } Map properties = configuration.getProperties(); @@ -140,7 +136,9 @@ public class ESAdapter implements OuterAdapter { String database = dml.getDatabase(); String table = dml.getTable(); Map configMap = dbTableEsSyncConfig.get(database + "-" + table); - esSyncService.sync(configMap.values(), dml); + if (configMap != null) { + esSyncService.sync(configMap.values(), dml); + } } @Override diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index e586bfe0..7377c43e 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -115,7 +115,9 @@ public class HbaseAdapter implements OuterAdapter { String database = dml.getDatabase(); String table = dml.getTable(); Map configMap = mappingConfigCache.get(destination + "." + database + "." + table); - configMap.values().forEach(config -> hbaseSyncService.sync(config, dml)); + if (configMap != null) { + configMap.values().forEach(config -> hbaseSyncService.sync(config, dml)); + } } @Override diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index 363d1410..b76b9687 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -2,11 +2,7 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -80,16 +76,17 @@ public abstract class AbstractCanalAdapterWorker { futures.forEach(future -> { try { if (!future.get()) { - logger.error("Outer adapter write failed"); + throw new RuntimeException("Outer adapter sync failed! "); } - } catch (InterruptedException | ExecutionException e) { - // ignore + } catch (Exception e) { + future.cancel(true); + throw new RuntimeException(e); } }); }); } - private void writeOut(final List flatMessages) { + protected void writeOut(final List flatMessages) { List> futures = new ArrayList<>(); // 组间适配器并行运行 canalOuterAdapters.forEach(outerAdapters -> { @@ -119,63 +116,64 @@ public abstract class AbstractCanalAdapterWorker { futures.forEach(future -> { try { if (!future.get()) { - logger.error("Outer adapter write failed"); + throw new RuntimeException("Outer adapter sync failed! "); } - } catch (InterruptedException | ExecutionException e) { - // ignore + } catch (Exception e) { + future.cancel(true); + throw new RuntimeException(e); } }); }); } @SuppressWarnings("unchecked") - protected void mqWriteOutData(int retry, long timeout, final boolean flatMessage, CanalMQConnector connector, + protected boolean mqWriteOutData(int retry, long timeout, int i, final boolean flatMessage, CanalMQConnector connector, ExecutorService workerExecutor) { - for (int i = 0; i < retry; i++) { - try { - List messages; - if (!flatMessage) { - messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS); - } else { - messages = connector.getFlatListWithoutAck(100L, TimeUnit.MILLISECONDS); - } - if (messages != null) { - Future future = workerExecutor.submit(() -> { - if (flatMessage) { - // batch write - writeOut((List) messages); - } else { - for (final Object message : messages) { - writeOut((Message) message); - } + try { + List messages; + if (!flatMessage) { + messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS); + } else { + messages = connector.getFlatListWithoutAck(100L, TimeUnit.MILLISECONDS); + } + if (messages != null && !messages.isEmpty()) { + Future future = workerExecutor.submit(() -> { + if (flatMessage) { + // batch write + writeOut((List) messages); + } else { + for (final Object message : messages) { + writeOut((Message) message); } - return true; - }); - - try { - future.get(timeout, TimeUnit.MILLISECONDS); - } catch (Exception e) { - future.cancel(true); - throw e; } + return true; + }); + + try { + future.get(timeout, TimeUnit.MILLISECONDS); + } catch (Exception e) { + future.cancel(true); + throw e; } connector.ack(); - break; - } catch (Throwable e) { - if (i == retry - 1) { - connector.ack(); - } else { - connector.rollback(); - } - - logger.error(e.getMessage(), e); - try { - TimeUnit.SECONDS.sleep(1L); - } catch (InterruptedException e1) { - // ignore - } + } + return true; + } catch (Throwable e) { + if (i == retry - 1) { + connector.ack(); + logger.error(e.getMessage() + " Error sync but ACK!"); + return true; + } else { + connector.rollback(); + logger.error(e.getMessage() + " Error sync and rollback, execute times: " + (i + 1)); + } + try { + Thread.sleep(500); + } catch (InterruptedException e1) { + // ignore } } + return false; } /** diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java index 050bbf65..ed83eb11 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -3,8 +3,11 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.protocol.Message; import org.apache.kafka.common.errors.WakeupException; import com.alibaba.otter.canal.client.adapter.OuterAdapter; @@ -61,7 +64,17 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { connector.disconnect(); break; } - mqWriteOutData(retry, timeout, flatMessage, connector, workerExecutor); + if (retry == -1) { + retry = Integer.MAX_VALUE; + } + for (int i = 0; i < retry; i++) { + if (!running) { + break; + } + if (mqWriteOutData(retry, timeout, i, flatMessage, connector, workerExecutor)) { + break; + } + } } } catch (Exception e) { logger.error(e.getMessage(), e); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java index 2e737353..806f37b6 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java @@ -3,7 +3,11 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.protocol.Message; import org.apache.kafka.common.errors.WakeupException; import com.alibaba.otter.canal.client.adapter.OuterAdapter; @@ -39,7 +43,8 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { ; ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); - int retry = canalClientConfig.getRetries() == null || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); + int retry = canalClientConfig.getRetries() == null + || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 while (running) { @@ -56,7 +61,17 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { connector.disconnect(); break; } - mqWriteOutData(retry, timeout, flatMessage, connector, workerExecutor); + if (retry == -1) { + retry = Integer.MAX_VALUE; + } + for (int i = 0; i < retry; i++) { + if (!running) { + break; + } + if (mqWriteOutData(retry, timeout, i, flatMessage, connector, workerExecutor)) { + break; + } + } } } catch (Exception e) { logger.error(e.getMessage(), e); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java index 15207d84..d04ab5a7 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java @@ -20,10 +20,10 @@ import com.alibaba.otter.canal.protocol.Message; */ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { - private static final int BATCH_SIZE = 50; - private static final int SO_TIMEOUT = 0; + private static final int BATCH_SIZE = 50; + private static final int SO_TIMEOUT = 0; - private CanalConnector connector; + private CanalConnector connector; /** * 单台client适配器worker的构造方法 @@ -61,10 +61,14 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { while (!running) ; // waiting until running == true - ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); int retry = canalClientConfig.getRetries() == null || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); - long timeout = canalClientConfig.getTimeout() == null ? 300000 : canalClientConfig.getTimeout(); // 默认超时5分钟 + if (retry == -1) { + // 重试次数-1代表异常时一直阻塞重试 + retry = Integer.MAX_VALUE; + } + // long timeout = canalClientConfig.getTimeout() == null ? 300000 : + // canalClientConfig.getTimeout(); // 默认超时5分钟 Integer batchSize = canalClientConfig.getBatchSize(); if (batchSize == null) { batchSize = BATCH_SIZE; @@ -90,6 +94,9 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } for (int i = 0; i < retry; i++) { + if (!running) { + break; + } Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据 long batchId = message.getId(); try { @@ -97,29 +104,19 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { if (batchId == -1 || size == 0) { Thread.sleep(500); } else { - Future future = workerExecutor.submit(() -> { - if (logger.isDebugEnabled()) { - logger.debug("destination: {} batchId: {} batchSize: {} ", - canalDestination, - batchId, - size); - } - long begin = System.currentTimeMillis(); - writeOut(message); - if (logger.isDebugEnabled()) { - logger.debug("destination: {} batchId: {} elapsed time: {} ms", - canalDestination, - batchId, - System.currentTimeMillis() - begin); - } - return true; - }); - - try { - future.get(timeout, TimeUnit.MILLISECONDS); - } catch (Exception e) { - future.cancel(true); - throw e; + if (logger.isDebugEnabled()) { + logger.debug("destination: {} batchId: {} batchSize: {} ", + canalDestination, + batchId, + size); + } + long begin = System.currentTimeMillis(); + writeOut(message); + if (logger.isDebugEnabled()) { + logger.debug("destination: {} batchId: {} elapsed time: {} ms", + canalDestination, + batchId, + System.currentTimeMillis() - begin); } } connector.ack(batchId); // 提交确认 @@ -127,10 +124,11 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } catch (Exception e) { if (i != retry - 1) { connector.rollback(batchId); // 处理失败, 回滚数据 + logger.error(e.getMessage() + " Error sync and rollback, execute times: " + (i + 1)); } else { connector.ack(batchId); + logger.error(e.getMessage() + " Error sync but ACK!"); } - logger.error("sync error!", e); Thread.sleep(500); } } @@ -151,8 +149,6 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } } } - - workerExecutor.shutdown(); } @Override 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 79f10955..989d0e5e 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 @@ -5,7 +5,10 @@ import java.sql.SQLException; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import javax.sql.DataSource; @@ -152,8 +155,8 @@ public class RdbAdapter implements OuterAdapter { try { future1.get(); future2.get(); - } catch (ExecutionException | InterruptedException e) { - // ignore + } catch (Exception e) { + throw new RuntimeException(e); } } 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 cd3c6567..2c94bee1 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 @@ -47,62 +47,58 @@ public class RdbMirrorDbSyncService { * @param dmls 批量 DML */ public void sync(List dmls) { - try { - List dmlList = new ArrayList<>(); - for (Dml dml : dmls) { - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); - MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(destination + "." + database); + List dmlList = new ArrayList<>(); + for (Dml dml : dmls) { + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(destination + "." + database); + if (mirrorDbConfig == null) { + continue; + } + if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { + // DDL + if (logger.isDebugEnabled()) { + logger.debug("DDL: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); + } + executeDdl(mirrorDbConfig, dml); + rdbSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable()); + mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置 + } else { + // DML + initMappingConfig(dml.getTable(), mirrorDbConfig.getMappingConfig(), mirrorDbConfig, dml); + dmlList.add(dml); + } + } + if (!dmlList.isEmpty()) { + rdbSyncService.sync(dmlList, dml -> { + MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(dml.getDestination() + "." + dml.getDatabase()); if (mirrorDbConfig == null) { - continue; + return false; } - if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { - // DDL - if (logger.isDebugEnabled()) { - logger.debug("DDL: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); - } - executeDdl(mirrorDbConfig, dml); - rdbSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable()); - mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置 + String table = dml.getTable(); + MappingConfig config = mirrorDbConfig.getTableConfig().get(table); + + if (config == null) { + return false; + } + + if (config.getConcurrent()) { + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData()); + RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); + rdbSyncService.getDmlsPartition()[hash].add(syncItem); + }); } else { - // DML - initMappingConfig(dml.getTable(), mirrorDbConfig.getMappingConfig(), mirrorDbConfig, dml); - dmlList.add(dml); + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); + rdbSyncService.getDmlsPartition()[hash].add(syncItem); + }); } - } - if (!dmlList.isEmpty()) { - rdbSyncService.sync(dmlList, dml -> { - MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache - .get(dml.getDestination() + "." + dml.getDatabase()); - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); - String table = dml.getTable(); - MappingConfig config = mirrorDbConfig.getTableConfig().get(table); - - if (config == null) { - return false; - } - - if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData()); - RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); - rdbSyncService.getDmlsPartition()[hash].add(syncItem); - }); - } else { - int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); - rdbSyncService.getDmlsPartition()[hash].add(syncItem); - }); - } - return true; - }); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); + return true; + }); } } 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 dd3a901a..6da01bf4 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 @@ -133,49 +133,44 @@ public class RdbSyncService { * @param dmls 批量 DML */ public void sync(Map> mappingConfig, List dmls) { - try { - sync(dmls, dml -> { - if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { - // DDL - columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); + sync(dmls, dml -> { + if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { + // DDL + columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); + return false; + } else { + // DML + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + String table = dml.getTable(); + Map configMap = mappingConfig.get(destination + "." + database + "." + table); + + if (configMap == null) { return false; - } else { - // DML - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); - String table = dml.getTable(); - Map configMap = mappingConfig - .get(destination + "." + database + "." + table); - - if (configMap == null) { - return false; - } - - boolean executed = false; - for (MappingConfig config : configMap.values()) { - if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - int hash = pkHash(config.getDbMapping(), singleDml.getData()); - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); - } else { - int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); - } - executed = true; - } - return executed; } - }); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } + + boolean executed = false; + for (MappingConfig config : configMap.values()) { + if (config.getConcurrent()) { + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + int hash = pkHash(config.getDbMapping(), singleDml.getData()); + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } else { + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } + executed = true; + } + return executed; + } + }); } /** diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java index 6e9b81a6..6be006d2 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SyncUtil.java @@ -86,9 +86,7 @@ public class SyncUtil { } break; case Types.TINYINT: - if (value instanceof Byte || value instanceof Short || value instanceof Integer) { - pstmt.setByte(i, (byte) value); - } else if (value instanceof Number) { + if (value instanceof Number) { pstmt.setByte(i, ((Number) value).byteValue()); } else if (value instanceof String) { pstmt.setByte(i, Byte.parseByte((String) value)); @@ -97,9 +95,7 @@ public class SyncUtil { } break; case Types.SMALLINT: - if (value instanceof Byte || value instanceof Short || value instanceof Integer) { - pstmt.setShort(i, (short) value); - } else if (value instanceof Number) { + if (value instanceof Number) { pstmt.setShort(i, ((Number) value).shortValue()); } else if (value instanceof String) { pstmt.setShort(i, Short.parseShort((String) value)); @@ -108,10 +104,7 @@ public class SyncUtil { } break; case Types.INTEGER: - if (value instanceof Byte || value instanceof Short || value instanceof Integer - || value instanceof Long) { - pstmt.setInt(i, (int) value); - } else if (value instanceof Number) { + if (value instanceof Number) { pstmt.setInt(i, ((Number) value).intValue()); } else if (value instanceof String) { pstmt.setInt(i, Integer.parseInt((String) value)); @@ -120,10 +113,7 @@ public class SyncUtil { } break; case Types.BIGINT: - if (value instanceof Byte || value instanceof Short || value instanceof Integer - || value instanceof Long) { - pstmt.setLong(i, (long) value); - } else if (value instanceof Number) { + if (value instanceof Number) { pstmt.setLong(i, ((Number) value).longValue()); } else if (value instanceof String) { pstmt.setLong(i, Long.parseLong((String) value)); @@ -136,13 +126,13 @@ public class SyncUtil { if (value instanceof BigDecimal) { pstmt.setBigDecimal(i, (BigDecimal) value); } else if (value instanceof Byte) { - pstmt.setInt(i, (int) value); + pstmt.setInt(i, ((Byte) value).intValue()); } else if (value instanceof Short) { - pstmt.setInt(i, (int) value); + pstmt.setInt(i, ((Short) value).intValue()); } else if (value instanceof Integer) { - pstmt.setInt(i, (int) value); + pstmt.setInt(i, (Integer) value); } else if (value instanceof Long) { - pstmt.setLong(i, (long) value); + pstmt.setLong(i, (Long) value); } else if (value instanceof Float) { pstmt.setBigDecimal(i, new BigDecimal((float) value)); } else if (value instanceof Double) { @@ -154,10 +144,7 @@ public class SyncUtil { } break; case Types.REAL: - if (value instanceof Byte || value instanceof Short || value instanceof Integer || value instanceof Long - || value instanceof Float || value instanceof Double) { - pstmt.setFloat(i, (float) value); - } else if (value instanceof Number) { + if (value instanceof Number) { pstmt.setFloat(i, ((Number) value).floatValue()); } else if (value instanceof String) { pstmt.setFloat(i, Float.parseFloat((String) value)); @@ -167,10 +154,7 @@ public class SyncUtil { break; case Types.FLOAT: case Types.DOUBLE: - if (value instanceof Byte || value instanceof Short || value instanceof Integer || value instanceof Long - || value instanceof Float || value instanceof Double) { - pstmt.setDouble(i, (double) value); - } else if (value instanceof Number) { + if (value instanceof Number) { pstmt.setDouble(i, ((Number) value).doubleValue()); } else if (value instanceof String) { pstmt.setDouble(i, Double.parseDouble((String) value)); diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java index 00403374..9b8b263a 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java @@ -1,9 +1,6 @@ package com.alibaba.otter.canal.client.kafka; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Properties; +import java.util.*; import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -22,7 +19,7 @@ import com.google.common.collect.Lists; /** * canal kafka 数据操作客户端 - * + * * 
  * 注意点：
  * 1. 相比于canal {@linkplain SimpleCanalConnector}, 这里get和ack操作不能有并发, 必须是一个线程执行get后，内存里执行完毕ack后再取下一个get
@@ -33,14 +30,16 @@ import com.google.common.collect.Lists;
  */
 public class KafkaCanalConnector implements CanalMQConnector {
 
-    private KafkaConsumer kafkaConsumer;
-    private KafkaConsumer  kafkaConsumer2;   // 用于扁平message的数据消费
-    private String                         topic;
-    private Integer                        partition;
-    private Properties                     properties;
-    private volatile boolean               connected = false;
-    private volatile boolean               running   = false;
-    private boolean                        flatMessage;
+    protected KafkaConsumer kafkaConsumer;
+    protected KafkaConsumer  kafkaConsumer2;                  // 用于扁平message的数据消费
+    protected String                         topic;
+    protected Integer                        partition;
+    protected Properties                     properties;
+    protected volatile boolean               connected      = false;
+    protected volatile boolean               running        = false;
+    protected boolean                        flatMessage;
+
+    private Map               currentOffsets = new HashMap<>();
 
     public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
                                boolean flatMessage){
@@ -71,6 +70,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
     /**
      * 打开连接
      */
+    @Override
     public void connect() {
         if (connected) {
             return;
@@ -79,6 +79,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
         connected = true;
         if (kafkaConsumer == null && !flatMessage) {
             kafkaConsumer = new KafkaConsumer(properties);
+
         }
         if (kafkaConsumer2 == null && flatMessage) {
             kafkaConsumer2 = new KafkaConsumer(properties);
@@ -88,6 +89,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
     /**
      * 关闭链接
      */
+    @Override
     public void disconnect() {
         if (kafkaConsumer != null) {
             kafkaConsumer.close();
@@ -101,10 +103,11 @@ public class KafkaCanalConnector implements CanalMQConnector {
         connected = false;
     }
 
-    private void waitClientRunning() {
+    protected void waitClientRunning() {
         running = true;
     }
 
+    @Override
     public boolean checkValid() {
         return true;// 默认都放过
     }
@@ -112,6 +115,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
     /**
      * 订阅topic
      */
+    @Override
     public void subscribe() {
         waitClientRunning();
         if (!running) {
@@ -139,6 +143,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
     /**
      * 取消订阅
      */
+    @Override
     public void unsubscribe() {
         waitClientRunning();
         if (!running) {
@@ -176,6 +181,11 @@ public class KafkaCanalConnector implements CanalMQConnector {
 
         ConsumerRecords records = kafkaConsumer.poll(unit.toMillis(timeout));
 
+        currentOffsets.clear();
+        for (TopicPartition topicPartition : records.partitions()) {
+            currentOffsets.put(topicPartition.partition(), kafkaConsumer.position(topicPartition));
+        }
+
         if (!records.isEmpty()) {
             List messages = new ArrayList<>();
             for (ConsumerRecord record : records) {
@@ -208,6 +218,12 @@ public class KafkaCanalConnector implements CanalMQConnector {
         }
 
         ConsumerRecords records = kafkaConsumer2.poll(unit.toMillis(timeout));
+
+        currentOffsets.clear();
+        for (TopicPartition topicPartition : records.partitions()) {
+            currentOffsets.put(topicPartition.partition(), kafkaConsumer2.position(topicPartition));
+        }
+
         if (!records.isEmpty()) {
             List flatMessages = new ArrayList<>();
             for (ConsumerRecord record : records) {
@@ -222,12 +238,28 @@ public class KafkaCanalConnector implements CanalMQConnector {
     }
 
     @Override
-    public void rollback() throws CanalClientException {
+    public void rollback() {
+        waitClientRunning();
+        if (!running) {
+            return;
+        }
+        // 回滚所有分区
+        if (kafkaConsumer != null) {
+            for (Map.Entry entry : currentOffsets.entrySet()) {
+                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
+            }
+        }
+        if (kafkaConsumer2 != null) {
+            for (Map.Entry entry : currentOffsets.entrySet()) {
+                kafkaConsumer2.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
+            }
+        }
     }
 
     /**
      * 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败
      */
+    @Override
     public void ack() {
         waitClientRunning();
         if (!running) {
diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaOffsetCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaOffsetCanalConnector.java
new file mode 100644
index 00000000..2bfa35f9
--- /dev/null
+++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaOffsetCanalConnector.java
@@ -0,0 +1,112 @@
+package com.alibaba.otter.canal.client.kafka;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.otter.canal.client.kafka.protocol.KafkaFlatMessage;
+import com.alibaba.otter.canal.client.kafka.protocol.KafkaMessage;
+import com.alibaba.otter.canal.protocol.FlatMessage;
+import com.alibaba.otter.canal.protocol.Message;
+import com.alibaba.otter.canal.protocol.exception.CanalClientException;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * kafka带消息offset的连接器
+ *
+ * @Author panjianping
+ * @Email ipanjianping@qq.com
+ * @Date 2018/12/17
+ */
+public class KafkaOffsetCanalConnector extends KafkaCanalConnector {
+
+    public KafkaOffsetCanalConnector(String servers, String topic, Integer partition, String groupId, boolean flatMessage) {
+        super(servers, topic, partition, groupId, 100, flatMessage);
+        // 启动时从未消费的消息位置开始
+        properties.put("auto.offset.reset", "earliest");
+    }
+
+    /**
+     * 获取Kafka消息，不确认
+     *
+     * @param timeout
+     * @param unit
+     * @param offset  消息偏移地址（-1为不偏移）
+     * @return
+     * @throws CanalClientException
+     */
+    public List getListWithoutAck(Long timeout, TimeUnit unit, long offset) throws CanalClientException {
+        waitClientRunning();
+        if (!running) {
+            return Lists.newArrayList();
+        }
+
+        if (offset > -1) {
+            TopicPartition tp = new TopicPartition(topic, partition == null ? 0 : partition);
+            kafkaConsumer.seek(tp, offset);
+        }
+
+        ConsumerRecords records = kafkaConsumer.poll(unit.toMillis(timeout));
+
+        if (!records.isEmpty()) {
+            List messages = new ArrayList<>();
+            for (ConsumerRecord record : records) {
+                KafkaMessage message = new KafkaMessage(record.value(), record.offset());
+                messages.add(message);
+            }
+            return messages;
+        }
+        return Lists.newArrayList();
+    }
+
+    /**
+     * 获取Kafka消息，不确认
+     *
+     * @param timeout
+     * @param unit
+     * @param offset  消息偏移地址（-1为不偏移）
+     * @return
+     * @throws CanalClientException
+     */
+    public List getFlatListWithoutAck(Long timeout, TimeUnit unit, long offset) throws CanalClientException {
+        waitClientRunning();
+        if (!running) {
+            return Lists.newArrayList();
+        }
+
+        if (offset > -1) {
+            TopicPartition tp = new TopicPartition(topic, partition == null ? 0 : partition);
+            kafkaConsumer2.seek(tp, offset);
+        }
+
+        ConsumerRecords records = kafkaConsumer2.poll(unit.toMillis(timeout));
+        if (!records.isEmpty()) {
+            List flatMessages = new ArrayList<>();
+            for (ConsumerRecord record : records) {
+                String flatMessageJson = record.value();
+                FlatMessage flatMessage = JSON.parseObject(flatMessageJson, FlatMessage.class);
+                KafkaFlatMessage message = new KafkaFlatMessage(flatMessage, record.offset());
+                flatMessages.add(message);
+            }
+
+            return flatMessages;
+        }
+        return Lists.newArrayList();
+    }
+
+    /**
+     * 重新设置AutoOffsetReset（默认 earliest ）
+     *
+     * @param value
+     */
+    public void setAutoOffsetReset(String value) {
+        if (StringUtils.isNotBlank(value)) {
+            properties.put("auto.offset.reset", value);
+        }
+    }
+}
diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/protocol/KafkaFlatMessage.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/protocol/KafkaFlatMessage.java
new file mode 100644
index 00000000..3661ec9e
--- /dev/null
+++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/protocol/KafkaFlatMessage.java
@@ -0,0 +1,32 @@
+package com.alibaba.otter.canal.client.kafka.protocol;
+
+import com.alibaba.otter.canal.protocol.FlatMessage;
+import org.springframework.beans.BeanUtils;
+
+/**
+ * 消息对象（Kafka）
+ *
+ * @Author panjianping
+ * @Email ipanjianping@qq.com
+ * @Date 2018/12/17
+ */
+public class KafkaFlatMessage extends FlatMessage {
+    /**
+     * Kafka 消息 offset
+     */
+    private long offset;
+
+    public KafkaFlatMessage(FlatMessage message, long offset) {
+        super(message.getId());
+        BeanUtils.copyProperties(message, this);
+        this.offset = offset;
+    }
+
+    public long getOffset() {
+        return offset;
+    }
+
+    public void setOffset(long offset) {
+        this.offset = offset;
+    }
+}
diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/protocol/KafkaMessage.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/protocol/KafkaMessage.java
new file mode 100644
index 00000000..f2f55a71
--- /dev/null
+++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/protocol/KafkaMessage.java
@@ -0,0 +1,33 @@
+package com.alibaba.otter.canal.client.kafka.protocol;
+
+import com.alibaba.otter.canal.protocol.Message;
+import org.springframework.beans.BeanUtils;
+
+/**
+ * 消息对象（Kafka）
+ *
+ * @Author panjianping
+ * @Email ipanjianping@qq.com
+ * @Date 2018/12/17
+ */
+public class KafkaMessage extends Message {
+    /**
+     * Kafka 消息 offset
+     */
+    private long offset;
+
+    public KafkaMessage(Message message, long offset) {
+        super(message.getId());
+        BeanUtils.copyProperties(message, this);
+        this.offset = offset;
+    }
+
+
+    public long getOffset() {
+        return offset;
+    }
+
+    public void setOffset(long offset) {
+        this.offset = offset;
+    }
+}
diff --git a/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaOffsetClientExample.java b/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaOffsetClientExample.java
new file mode 100644
index 00000000..aaf367d4
--- /dev/null
+++ b/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaOffsetClientExample.java
@@ -0,0 +1,168 @@
+package com.alibaba.otter.canal.client.running.kafka;
+
+import com.alibaba.otter.canal.client.kafka.KafkaOffsetCanalConnector;
+import com.alibaba.otter.canal.client.kafka.protocol.KafkaMessage;
+import org.apache.kafka.common.errors.WakeupException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.util.Assert;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * KafkaOffsetCanalConnector 使用示例
+ * 
KafkaOffsetCanalConnector 与 KafkaCanalConnector 的另一区别是 auto.offset.reset 默认值不同；

+ * 
KafkaOffsetCanalConnector 默认为 earliest；canal-kafka-client重启后从未被消费的记录开始拉取消息，同时提供了修改 auto.offset.reset 的方法 setAutoOffsetReset

+ *
+ * @author panjianping @ 2018-12-18
+ * @version 1.1.3
+ */
+public class CanalKafkaOffsetClientExample {
+
+    protected final static Logger logger = LoggerFactory.getLogger(CanalKafkaOffsetClientExample.class);
+
+    private KafkaOffsetCanalConnector connector;
+
+    private static volatile boolean running = false;
+
+    private Thread thread = null;
+
+    private Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
+
+        public void uncaughtException(Thread t, Throwable e) {
+            logger.error("parse events has an error", e);
+        }
+    };
+
+    public CanalKafkaOffsetClientExample(String servers, String topic, Integer partition, String groupId) {
+        connector = new KafkaOffsetCanalConnector(servers, topic, partition, groupId, false);
+    }
+
+    public static void main(String[] args) {
+        try {
+            final CanalKafkaOffsetClientExample kafkaCanalClientExample = new CanalKafkaOffsetClientExample(
+                    AbstractKafkaTest.servers,
+                    AbstractKafkaTest.topic,
+                    AbstractKafkaTest.partition,
+                    AbstractKafkaTest.groupId);
+            logger.info("## start the kafka consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId);
+            kafkaCanalClientExample.start();
+            logger.info("## the canal kafka consumer is running now ......");
+            Runtime.getRuntime().addShutdownHook(new Thread() {
+
+                public void run() {
+                    try {
+                        logger.info("## stop the kafka consumer");
+                        kafkaCanalClientExample.stop();
+                    } catch (Throwable e) {
+                        logger.warn("##something goes wrong when stopping kafka consumer:", e);
+                    } finally {
+                        logger.info("## kafka consumer is down.");
+                    }
+                }
+
+            });
+            while (running)
+                ;
+        } catch (Throwable e) {
+            logger.error("## Something goes wrong when starting up the kafka consumer:", e);
+            System.exit(0);
+        }
+    }
+
+    public void start() {
+        Assert.notNull(connector, "connector is null");
+        thread = new Thread(new Runnable() {
+
+            public void run() {
+                process();
+            }
+        });
+        thread.setUncaughtExceptionHandler(handler);
+        thread.start();
+        running = true;
+    }
+
+    public void stop() {
+        if (!running) {
+            return;
+        }
+        running = false;
+        if (thread != null) {
+            try {
+                thread.join();
+            } catch (InterruptedException e) {
+                // ignore
+            }
+        }
+    }
+
+    private void process() {
+        while (!running)
+            ;
+        while (running) {
+            try {
+                // 修改 AutoOffsetReset 的值，默认（earliest）
+                //connector.setAutoOffsetReset(null);
+                connector.connect();
+                connector.subscribe();
+                // 消息起始偏移地址
+                long offset = -1;
+                // 错误次数
+                int errorCount = 0;
+                while (running) {
+                    try {
+                        // 错误重试次数超过3次后，每30秒递增重试
+                        if (errorCount > 2) {
+                            Thread.sleep((errorCount - 2) * 1000 * 30);
+                        }
+
+                        List messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS, offset); // 获取message
+                        if (messages == null) {
+                            continue;
+                        }
+                        for (KafkaMessage message : messages) {
+                            long batchId = message.getId();
+                            int size = message.getEntries().size();
+
+                            if (batchId == -1 || size == 0) {
+                                continue;
+                            }
+
+                            // 记录第一条消息的offset，用于处理数据异常时重新从此位置获取消息
+                            if (offset < 0) {
+                                offset = message.getOffset();
+                            }
+
+                            // printSummary(message, batchId, size);
+                            // printEntry(message.getEntries());
+                            logger.info(message.toString());
+                        }
+
+                        connector.ack(); // 提交确认
+                        // 还原offset
+                        offset = -1;
+                        errorCount = 0;
+                    } catch (Exception e) {
+                        errorCount++;
+                        logger.error(e.getMessage(), e);
+                        if (errorCount == 3) {
+                            // 重试3次后发送邮件提醒异常
+                            // mailService.sendMail("同步数据异常，请及时处理", "错误消息");
+                        }
+                    }
+                }
+            } catch (Exception e) {
+                logger.error(e.getMessage(), e);
+            }
+        }
+
+        try {
+            connector.unsubscribe();
+        } catch (WakeupException e) {
+            // No-op. Continue process
+        }
+        connector.disconnect();
+    }
+}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
index 0a0f6314..f4a23dc8 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
@@ -30,6 +30,7 @@ import com.alibaba.fastsql.sql.ast.statement.SQLSelectOrderByItem;
 import com.alibaba.fastsql.sql.ast.statement.SQLTableElement;
 import com.alibaba.fastsql.sql.dialect.mysql.ast.MySqlPrimaryKey;
 import com.alibaba.fastsql.sql.dialect.mysql.ast.MySqlUnique;
+import com.alibaba.fastsql.sql.dialect.mysql.ast.expr.MySqlOrderingExpr;
 import com.alibaba.fastsql.sql.repository.Schema;
 import com.alibaba.fastsql.sql.repository.SchemaObject;
 import com.alibaba.fastsql.sql.repository.SchemaRepository;
@@ -258,6 +259,8 @@ public class MemoryTableMeta implements TableMetaTSDB {
             return ((SQLCharExpr) sqlName).getText();
         } else if (sqlName instanceof SQLMethodInvokeExpr) {
             return DruidDdlParser.unescapeName(((SQLMethodInvokeExpr) sqlName).getMethodName());
+        } else if (sqlName instanceof MySqlOrderingExpr) {
+            return getSqlName(((MySqlOrderingExpr) sqlName).getExpr());
         } else {
             return sqlName.toString();
         }