From b2b69307f9fbd1fe483ebe498883613531898f36 Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 21 Mar 2019 20:50:52 +0800 Subject: [PATCH] fixed rocketmq --- .../otter/canal/client/rocketmq/RocketMQCanalConnector.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java index 59b10c1b..a39c1ba5 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java @@ -12,7 +12,6 @@ import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely; import org.apache.rocketmq.client.exception.MQClientException; -import org.apache.rocketmq.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.remoting.RPCHook; import org.slf4j.Logger; @@ -84,7 +83,6 @@ public class RocketMQCanalConnector implements CanalMQConnector { } rocketMQConsumer = new DefaultMQPushConsumer(groupName, rpcHook, new AllocateMessageQueueAveragely()); rocketMQConsumer.setVipChannelEnabled(false); - rocketMQConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); if (!StringUtils.isBlank(nameServer)) { rocketMQConsumer.setNamesrvAddr(nameServer); }