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 #1420

Browse Source
This commit is contained in:
mcy
2019-01-15 11:38:00 +08:00
parent 18b304f99d
commit ca41be8412
1 changed files with 13 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -144,7 +144,10 @@ public class CanalKafkaProducer implements CanalMQProducer {
// 发送扁平数据json // 发送扁平数据json
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(message); List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(message);
if (flatMessages != null) { if (flatMessages != null) {
int idx = 0;
int size = flatMessages.size();
for (FlatMessage flatMessage : flatMessages) { for (FlatMessage flatMessage : flatMessages) {
idx++;
if (StringUtils.isEmpty(canalDestination.getPartitionHash())) { if (StringUtils.isEmpty(canalDestination.getPartitionHash())) {
try { try {
Integer partition = canalDestination.getPartition(); Integer partition = canalDestination.getPartition();
@@ -155,7 +158,11 @@ public class CanalKafkaProducer implements CanalMQProducer {
partition, partition,
null, null,
JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue));
producer2.send(record); if (idx != size) {
producer2.send(record);
} else {
producer2.send(record).get();
}
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
// producer.abortTransaction(); // producer.abortTransaction();
@@ -175,7 +182,11 @@ public class CanalKafkaProducer implements CanalMQProducer {
i, i,
null, null,
JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue)); JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue));
producer2.send(record); if (idx != size) {
producer2.send(record);
} else {
producer2.send(record).get();
}
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
// producer.abortTransaction(); // producer.abortTransaction();