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

fix序列化的一些问题, kafka生产端统一使用异步

Browse Source
This commit is contained in:
mcy
2018-09-17 11:10:50 +08:00
parent db56effd91
commit 7005c11ac2
3 changed files with 11 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java
View File

@@ -180,8 +180,8 @@ public class SimpleCanalConnector implements CanalConnector {
Ack ackBody = Ack.parseFrom(ack.getBody()); Ack ackBody = Ack.parseFrom(ack.getBody());
if (ackBody.getErrorCode() > 0) { if (ackBody.getErrorCode() > 0) {
throw new CanalClientException("something goes wrong when doing authentication: " throw new CanalClientException(
+ ackBody.getErrorMessage()); "something goes wrong when doing authentication: " + ackBody.getErrorMessage());
} }
connected = true; connected = true;
@@ -323,7 +323,9 @@ public class SimpleCanalConnector implements CanalConnector {
Packet p = Packet.parseFrom(data); Packet p = Packet.parseFrom(data);
switch (p.getType()) { switch (p.getType()) {
case MESSAGES: { case MESSAGES: {
if (!p.getCompression().equals(Compression.NONE)) { // if (!p.getCompression().equals(Compression.NONE)) {
if (!p.getCompression().equals(Compression.NONE)
&& !p.getCompression().equals(Compression.COMPRESSIONCOMPATIBLEPROTO2)) {
throw new CanalClientException("compression is not supported in this connector"); throw new CanalClientException("compression is not supported in this connector");
} }
@@ -360,11 +362,8 @@ public class SimpleCanalConnector implements CanalConnector {
.setBatchId(batchId) .setBatchId(batchId)
.build(); .build();
try { try {
writeWithHeader(Packet.newBuilder() writeWithHeader(
.setType(PacketType.CLIENTACK) Packet.newBuilder().setType(PacketType.CLIENTACK).setBody(ca.toByteString()).build().toByteArray());
.setBody(ca.toByteString())
.build()
.toByteArray());
} catch (IOException e) { } catch (IOException e) {
throw new CanalClientException(e); throw new CanalClientException(e);
} }

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

@@ -36,7 +36,8 @@ public class MessageDeserializer implements Deserializer<Message> {
CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data); CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data);
switch (p.getType()) { switch (p.getType()) {
case MESSAGES: { case MESSAGES: {
if (!p.getCompression().equals(CanalPacket.Compression.NONE)) { if (!p.getCompression().equals(CanalPacket.Compression.NONE)
&& !p.getCompression().equals(CanalPacket.Compression.COMPRESSIONCOMPATIBLEPROTO2)) {
throw new CanalClientException("compression is not supported in this connector"); throw new CanalClientException("compression is not supported in this connector");
} }

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

@@ -79,8 +79,8 @@ public class CanalKafkaProducer {
} else { } else {
record = new ProducerRecord<String, Message>(topic.getTopic(), message); record = new ProducerRecord<String, Message>(topic.getTopic(), message);
} }
Future<RecordMetadata> future = producer.send(record);
future.get(); producer.send(record);
} else { } else {
// 发送扁平数据json // 发送扁平数据json
List<FlatMessage> flatMessages = FlatMessage.messageConverter(message); List<FlatMessage> flatMessages = FlatMessage.messageConverter(message);