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

fixed test

Browse Source
This commit is contained in:
七锋
2018-09-30 11:33:48 +08:00
parent 9344ed121f
commit eb2a83b975
6 changed files with 48 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
View File

@@ -13,7 +13,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
@@ -50,7 +49,7 @@ public class KafkaCanalConnector {
properties.put("max.poll.records", "100");
properties.put("key.deserializer", StringDeserializer.class.getName());
if (!flatMessage) {
properties.put("value.deserializer", CanalMessageDeserializer.class.getName());
properties.put("value.deserializer", MessageDeserializer.class.getName());
} else {
properties.put("value.deserializer", StringDeserializer.class.getName());
}

31
client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java Normal file
View File

@@ -0,0 +1,31 @@
package com.alibaba.otter.canal.client.kafka;
import java.util.Map;
import org.apache.kafka.common.serialization.Deserializer;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.protocol.Message;
/**
* Kafka Message类的反序列化
*
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
public class MessageDeserializer implements Deserializer<Message> {
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
}
@Override
public Message deserialize(String topic1, byte[] data) {
return CanalMessageDeserializer.deserializer(data);
}
@Override
public void close() {
// nothing to do
}
}