From b1e2ab444807c2b8af28bfb97515646405a402f0 Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 23 Oct 2018 23:25:05 +0800 Subject: [PATCH 1/3] Update README.md --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 9baf3556..c371fe05 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ canal 1.1.x系列，参考release文档：wiki文 QuickStart From 80d3781d8192db8346b4e8009701dc6fc194bf61 Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 23 Oct 2018 23:27:42 +0800 Subject: [PATCH 2/3] Update README.md --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index c371fe05..6af316ae 100644 --- a/README.md +++ b/README.md @@ -85,11 +85,12 @@ See the wiki page for : wiki文 * canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp * canal go客户端开源项目地址： https://github.com/CanalClient/canal-go 2. canal作为MySQL binlog的增量获取工具，可以将数据投递到MQ系统中，比如Kafka/RocketMQ，可以借助于MQ的多语言能力 + * 参考文档: [Canal Kafka/RocketMQ QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart)

相关资料

* ADC阿里技术嘉年华分享ppt (放在google docs上，可能需要翻墙): ppt下载 -* [与阿里巴巴的RocketMQ配合使用](https://github.com/alibaba/RocketMQ) +* [与阿里巴巴的RocketMQ配合使用](https://github.com/apache/RocketMQ)

相关开源

    From 5bc5eec85e99e3be0f068a556e7cb4d4be2647a0 Mon Sep 17 00:00:00 2001 From: "titeng.jiang" Date: Wed, 24 Oct 2018 10:02:50 +0800 Subject: [PATCH 3/3] Fix Kafka FlatMessage Model Null Value set to "" --- .../alibaba/otter/canal/protocol/FlatMessage.java | 12 ++++++++++-- .../otter/canal/kafka/CanalKafkaProducer.java | 3 ++- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index f66cf9b5..dfa4280b 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -204,7 +204,11 @@ public class FlatMessage implements Serializable { for (CanalEntry.Column column : columns) { sqlType.put(column.getName(), column.getSqlType()); mysqlType.put(column.getName(), column.getMysqlType()); - row.put(column.getName(), column.getValue()); + if (column.getIsNull()) { + row.put(column.getName(), null); + } else { + row.put(column.getName(), column.getValue()); + } // 获取update为true的字段 if (column.getUpdated()) { updateSet.add(column.getName()); @@ -218,7 +222,11 @@ public class FlatMessage implements Serializable { Map rowOld = new LinkedHashMap<>(); for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { if (updateSet.contains(column.getName())) { - rowOld.put(column.getName(), column.getValue()); + if (column.getIsNull()) { + rowOld.put(column.getName(), null); + } else { + rowOld.put(column.getName(), column.getValue()); + } } } // update操作将记录修改前的值 diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 15217f13..93f5f4ab 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -5,6 +5,7 @@ import java.util.List; import java.util.Properties; import java.util.concurrent.Future; +import com.alibaba.fastjson.serializer.SerializerFeature; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; @@ -141,7 +142,7 @@ public class CanalKafkaProducer implements CanalMQProducer { canalDestination.getTopic(), 0, null, - JSON.toJSONString(flatMessage)); + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); producer2.send(record).get(); } catch (Exception e) { logger.error(e.getMessage(), e);