From ef72cddfe8c576dcb6aab7acf6141e86e7cc4978 Mon Sep 17 00:00:00 2001 From: panjianping <601738327@qq.com> Date: Tue, 18 Dec 2018 11:54:41 +0800 Subject: [PATCH 1/8] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=B8=A6kafka=E6=B6=88?= =?UTF-8?q?=E6=81=AFoffset=E7=9A=84Connector?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/kafka/KafkaCanalConnector.java | 24 ++-- .../kafka/KafkaOffsetCanalConnector.java | 112 ++++++++++++++++++ .../kafka/protocol/KafkaFlatMessage.java | 32 +++++ .../client/kafka/protocol/KafkaMessage.java | 33 ++++++ 4 files changed, 192 insertions(+), 9 deletions(-) create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaOffsetCanalConnector.java create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/kafka/protocol/KafkaFlatMessage.java create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/kafka/protocol/KafkaMessage.java 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..9248738c 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 @@ -33,14 +33,14 @@ 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; + private volatile boolean connected = false; + protected volatile boolean running = false; + private boolean flatMessage; public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize, boolean flatMessage){ @@ -71,6 +71,7 @@ public class KafkaCanalConnector implements CanalMQConnector { /** * 打开连接 */ + @Override public void connect() { if (connected) { return; @@ -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) { @@ -228,6 +233,7 @@ public class KafkaCanalConnector implements CanalMQConnector { /** * 提交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..6065f78c --- /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", "earliest"); + } + } +} 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; + } +} From 990fa9498a039a8b2b510f8d3d96ca149ede6b8e Mon Sep 17 00:00:00 2001 From: panjianping <601738327@qq.com> Date: Tue, 18 Dec 2018 16:46:19 +0800 Subject: [PATCH 2/8] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=20KafkaOffsetCanalConnec?= =?UTF-8?q?tor=20=E4=BD=BF=E7=94=A8=E7=A4=BA=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../kafka/KafkaOffsetCanalConnector.java | 2 +- .../kafka/CanalKafkaOffsetClientExample.java | 168 ++++++++++++++++++ 2 files changed, 169 insertions(+), 1 deletion(-) create mode 100644 client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaOffsetClientExample.java 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 index 6065f78c..2bfa35f9 100644 --- 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 @@ -106,7 +106,7 @@ public class KafkaOffsetCanalConnector extends KafkaCanalConnector { */ public void setAutoOffsetReset(String value) { if (StringUtils.isNotBlank(value)) { - properties.put("auto.offset.reset", "earliest"); + properties.put("auto.offset.reset", value); } } } 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(); + } +} From 035ef265426cf741b5db606ce9918eb09b945052 Mon Sep 17 00:00:00 2001 From: agapple Date: Wed, 19 Dec 2018 11:43:49 +0800 Subject: [PATCH 3/8] fixed issue #1306 , support MySqlOrderingExpr --- .../otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java | 3 +++ 1 file changed, 3 insertions(+) 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(); } From 75ee65f9e8a7661be3cb022473aa8a6a97262cac Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Dec 2018 19:45:57 +0800 Subject: [PATCH 4/8] =?UTF-8?q?=E7=B1=BB=E5=9E=8B=E8=BD=AC=E6=8D=A2?= =?UTF-8?q?=E9=97=AE=E9=A2=98=20fixed=20#1313?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../rdb/service/RdbMirrorDbSyncService.java | 5 +-- .../client/adapter/rdb/support/SyncUtil.java | 36 ++++++------------- 2 files changed, 13 insertions(+), 28 deletions(-) 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..9cc1e7db 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 @@ -74,8 +74,9 @@ public class RdbMirrorDbSyncService { rdbSyncService.sync(dmlList, dml -> { MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache .get(dml.getDestination() + "." + dml.getDatabase()); - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); + if (mirrorDbConfig == null) { + return false; + } String table = dml.getTable(); MappingConfig config = mirrorDbConfig.getTableConfig().get(table); 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..e8789b03 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); } else if (value instanceof Short) { - pstmt.setInt(i, (int) value); + pstmt.setInt(i, (Short) value); } 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)); From 34b5f9c927c41d18b780a0fc0c60bca2be8a55bb Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Dec 2018 19:53:30 +0800 Subject: [PATCH 5/8] modify --- .../otter/canal/client/adapter/rdb/support/SyncUtil.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 e8789b03..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 @@ -126,9 +126,9 @@ public class SyncUtil { if (value instanceof BigDecimal) { pstmt.setBigDecimal(i, (BigDecimal) value); } else if (value instanceof Byte) { - pstmt.setInt(i, (Byte) value); + pstmt.setInt(i, ((Byte) value).intValue()); } else if (value instanceof Short) { - pstmt.setInt(i, (Short) value); + pstmt.setInt(i, ((Short) value).intValue()); } else if (value instanceof Integer) { pstmt.setInt(i, (Integer) value); } else if (value instanceof Long) { From 354d64069f2455885eb9b6d8208b5f22ce1efe64 Mon Sep 17 00:00:00 2001 From: mcy Date: Sun, 23 Dec 2018 18:16:47 +0800 Subject: [PATCH 6/8] =?UTF-8?q?adapter=E5=A2=9E=E5=8A=A0=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E5=BC=82=E5=B8=B8=E9=87=8D=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../loader/AbstractCanalAdapterWorker.java | 102 +++++++++--------- .../loader/CanalAdapterKafkaWorker.java | 15 ++- .../loader/CanalAdapterRocketMQWorker.java | 19 +++- .../launcher/loader/CanalAdapterWorker.java | 58 +++++----- .../canal/client/adapter/rdb/RdbAdapter.java | 9 +- .../rdb/service/RdbMirrorDbSyncService.java | 101 +++++++++-------- .../adapter/rdb/service/RdbSyncService.java | 77 +++++++------ .../client/kafka/KafkaCanalConnector.java | 44 ++++++-- 8 files changed, 233 insertions(+), 192 deletions(-) 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 9cc1e7db..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,63 +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()); - if (mirrorDbConfig == null) { - return false; - } - 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/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..bedaca04 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
@@ -34,14 +31,16 @@ import com.google.common.collect.Lists;
 public class KafkaCanalConnector implements CanalMQConnector {
 
     private KafkaConsumer kafkaConsumer;
-    private KafkaConsumer  kafkaConsumer2;   // 用于扁平message的数据消费
+    private KafkaConsumer  kafkaConsumer2;                   // 用于扁平message的数据消费
     private String                         topic;
     private Integer                        partition;
     private Properties                     properties;
-    private volatile boolean               connected = false;
-    private volatile boolean               running   = false;
+    private volatile boolean               connected       = false;
+    private volatile boolean               running         = false;
     private boolean                        flatMessage;
 
+    private Map             currentOffsets  = new HashMap<>();
+
     public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
                                boolean flatMessage){
         this.topic = topic;
@@ -79,6 +78,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
         connected = true;
         if (kafkaConsumer == null && !flatMessage) {
             kafkaConsumer = new KafkaConsumer(properties);
+
         }
         if (kafkaConsumer2 == null && flatMessage) {
             kafkaConsumer2 = new KafkaConsumer(properties);
@@ -176,6 +176,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 +213,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,7 +233,22 @@ 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);
+            }
+        }
     }
 
     /**

From b5697a366abcf0c03e5a58d879740ac0f196dd71 Mon Sep 17 00:00:00 2001
From: mcy 
Date: Sun, 23 Dec 2018 18:24:04 +0800
Subject: [PATCH 7/8] merge

---
 .../client/kafka/KafkaCanalConnector.java      | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

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 5e9c99c2..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
@@ -30,16 +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<>();
+    private Map               currentOffsets = new HashMap<>();
 
     public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
                                boolean flatMessage){

From dffbc437a6f9204c1a3bf6b4304ea45e6aa3fa62 Mon Sep 17 00:00:00 2001
From: machey 
Date: Tue, 25 Dec 2018 22:27:27 +0800
Subject: [PATCH 8/8] fixed #1338

---
 .../canal/client/adapter/es/ESAdapter.java    | 22 +++++++++----------
 .../client/adapter/hbase/HbaseAdapter.java    |  4 +++-
 2 files changed, 13 insertions(+), 13 deletions(-)

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