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

Browse Source
This commit is contained in:
mcy
2018-11-01 14:46:29 +08:00
parent ed1feb9502
commit b9c4884af1
1 changed files with 26 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -77,8 +77,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
@@ -98,6 +98,7 @@ 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];
if (flatMessagePart != null) {
logger.debug("flatMessagePart: {}, partition: {}", logger.debug("flatMessagePart: {}, partition: {}",
JSON.toJSONString(flatMessagePart), JSON.toJSONString(flatMessagePart),
i); i);
@@ -108,9 +109,11 @@ public class CanalRocketMQProducer implements CanalMQProducer {
this.defaultMQProducer.send(message, new MessageQueueSelector() { 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,
Object arg) {
if (index > mqs.size()) { if (index > mqs.size()) {
throw new CanalServerException("partition number is error,config num:" throw new CanalServerException(
"partition number is error,config num:"
+ destination.getPartitionsNum() + destination.getPartitionsNum()
+ ", mq num: " + mqs.size()); + ", mq num: " + mqs.size());
} }
@@ -127,6 +130,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
} }
} }
} }
}
callback.commit(); callback.commit();
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {