chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

flagMessage json 序列化无null值的bug

Browse Source
This commit is contained in:
mcy
2018-11-08 11:22:34 +08:00
parent b71e299161
commit 6fa111866f
2 changed files with 7 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -104,7 +104,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination.getTopic(), ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination.getTopic(),
canalDestination.getPartition(), canalDestination.getPartition(),
null, null,
JSON.toJSONString(flatMessage)); JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue));
producer2.send(record).get(); producer2.send(record).get();
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
@@ -126,7 +126,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination.getTopic(), ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination.getTopic(),
i, i,
null, null,
JSON.toJSONString(flatMessagePart)); JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue));
producer2.send(record).get(); producer2.send(record).get();
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);

9
server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
View File

@@ -1,6 +1,7 @@
package com.alibaba.otter.canal.rocketmq; package com.alibaba.otter.canal.rocketmq;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.otter.canal.common.CanalMessageSerializer; import com.alibaba.otter.canal.common.CanalMessageSerializer;
import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.FlatMessage;
@@ -75,11 +76,11 @@ public class CanalRocketMQProducer implements CanalMQProducer {
if (destination.getPartition() != null) { if (destination.getPartition() != null) {
try { try {
logger.info("send flat message: {} to topic: {} fixed partition: {}", logger.info("send flat message: {} to topic: {} fixed partition: {}",
JSON.toJSONString(flatMessage), JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue),
destination.getTopic(), destination.getTopic(),
destination.getPartition()); destination.getPartition());
Message message = new Message(destination.getTopic(), Message message = new Message(destination.getTopic(),
JSON.toJSONString(flatMessage).getBytes()); JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue).getBytes());
this.defaultMQProducer.send(message, new MessageQueueSelector() { this.defaultMQProducer.send(message, new MessageQueueSelector() {
@Override @Override
@@ -102,12 +103,12 @@ public class CanalRocketMQProducer implements CanalMQProducer {
FlatMessage flatMessagePart = partitionFlatMessage[i]; FlatMessage flatMessagePart = partitionFlatMessage[i];
if (flatMessagePart != null) { if (flatMessagePart != null) {
logger.debug("flatMessagePart: {}, partition: {}", logger.debug("flatMessagePart: {}, partition: {}",
JSON.toJSONString(flatMessagePart), JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue),
i); i);
final int index = i; final int index = i;
try { try {
Message message = new Message(destination.getTopic(), Message message = new Message(destination.getTopic(),
JSON.toJSONString(flatMessagePart).getBytes()); JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue).getBytes());
this.defaultMQProducer.send(message, new MessageQueueSelector() { this.defaultMQProducer.send(message, new MessageQueueSelector() {
@Override @Override