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

Merge pull request #1084 from rewerma/master

Browse Source 
fix #1083
This commit is contained in:
rewerma
2018-11-01 14:47:33 +08:00
committed by GitHub
parent 0dea4b7944 b9c4884af1
commit d3e1ddd234
1 changed files with 26 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -78,8 +78,8 @@ public class CanalRocketMQProducer implements CanalMQProducer {
JSON.toJSONString(flatMessage), JSON.toJSONString(flatMessage),
destination.getTopic(), destination.getTopic(),
destination.getPartition()); destination.getPartition());
Message message = new Message(destination.getTopic(), JSON.toJSONString(flatMessage) Message message = new Message(destination.getTopic(),
.getBytes()); JSON.toJSONString(flatMessage).getBytes());
this.defaultMQProducer.send(message, new MessageQueueSelector() { this.defaultMQProducer.send(message, new MessageQueueSelector() {
@Override @Override
@@ -99,28 +99,32 @@ public class CanalRocketMQProducer implements CanalMQProducer {
int length = partitionFlatMessage.length; int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
FlatMessage flatMessagePart = partitionFlatMessage[i]; FlatMessage flatMessagePart = partitionFlatMessage[i];
logger.debug("flatMessagePart: {}, partition: {}", if (flatMessagePart != null) {
JSON.toJSONString(flatMessagePart), logger.debug("flatMessagePart: {}, partition: {}",
i); JSON.toJSONString(flatMessagePart),
final int index = i; i);
try { final int index = i;
Message message = new Message(destination.getTopic(), try {
JSON.toJSONString(flatMessagePart).getBytes()); Message message = new Message(destination.getTopic(),
this.defaultMQProducer.send(message, new MessageQueueSelector() { JSON.toJSONString(flatMessagePart).getBytes());
this.defaultMQProducer.send(message, new MessageQueueSelector() {
@Override @Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { public MessageQueue select(List<MessageQueue> mqs, Message msg,
if (index > mqs.size()) { Object arg) {
throw new CanalServerException("partition number is error,config num:" if (index > mqs.size()) {
+ destination.getPartitionsNum() throw new CanalServerException(
+ ", mq num: " + mqs.size()); "partition number is error,config num:"
+ destination.getPartitionsNum()
+ ", mq num: " + mqs.size());
}
return mqs.get(index);
} }
return mqs.get(index); }, null);
} } catch (Exception e) {
}, null); logger.error("send flat message to hashed partition error", e);
} catch (Exception e) { callback.rollback();
logger.error("send flat message to hashed partition error", e); }
callback.rollback();
} }
} }
} }