合并代码，解决冲突

Author: vintagewang

rocketmq-client/src/main/java/com/alibaba/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java

@@ -56,6 +56,7 @@
 import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
 import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
 import com.alibaba.rocketmq.common.sysflag.PullSysFlag;
+import com.alibaba.rocketmq.remoting.RPCHook;
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
 
 
@@ -79,9 +80,12 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
     // Consumer启动时间
     private final long consumerStartTimestamp = System.currentTimeMillis();
 
+    private final RPCHook rpcHook;
 
-    public DefaultMQPullConsumerImpl(final DefaultMQPullConsumer defaultMQPullConsumer) {
+
+    public DefaultMQPullConsumerImpl(final DefaultMQPullConsumer defaultMQPullConsumer, final RPCHook rpcHook) {
         this.defaultMQPullConsumer = defaultMQPullConsumer;
+        this.rpcHook = rpcHook;
     }
 
 
@@ -497,7 +501,8 @@ public void start() throws MQClientException {
             }
 
             this.mQClientFactory =
-                    MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPullConsumer);
+                    MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPullConsumer,
+                        this.rpcHook);
 
             // 初始化Rebalance变量
             this.rebalanceImpl.setConsumerGroup(this.defaultMQPullConsumer.getConsumerGroup());

rocketmq-client/src/main/java/com/alibaba/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java

@@ -68,6 +68,7 @@
 import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
 import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
 import com.alibaba.rocketmq.common.sysflag.PullSysFlag;
+import com.alibaba.rocketmq.remoting.RPCHook;
 import com.alibaba.rocketmq.remoting.exception.RemotingException;
 
 
@@ -123,9 +124,12 @@ public void registerFilterMessageHook(final FilterMessageHook hook) {
      */
     private final ArrayList<ConsumeMessageHook> consumeMessageHookList = new ArrayList<ConsumeMessageHook>();
 
+    private final RPCHook rpcHook;
 
-    public DefaultMQPushConsumerImpl(DefaultMQPushConsumer defaultMQPushConsumer) {
+
+    public DefaultMQPushConsumerImpl(DefaultMQPushConsumer defaultMQPushConsumer, RPCHook rpcHook) {
         this.defaultMQPushConsumer = defaultMQPushConsumer;
+        this.rpcHook = rpcHook;
     }
 
 
@@ -658,7 +662,8 @@ public void start() throws MQClientException {
             }
 
             this.mQClientFactory =
-                    MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer);
+                    MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer,
+                        this.rpcHook);
 
             // 初始化Rebalance变量
             this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());