Merge 18cb81b into 98bd032

apache · Aug 10, 2017 · 96506c9 · 96506c9
2 parents 98bd032 + 18cb81b
commit 96506c9
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 0 deletions.
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
@@ -569,6 +569,7 @@ public synchronized void start() throws MQClientException {
                         default:
                             break;
                     }
+                    this.defaultMQPullConsumer.setOffsetStore(this.offsetStore);
                 }
 
                 this.offsetStore.load();

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
@@ -577,6 +577,7 @@ public synchronized void start() throws MQClientException {
                         default:
                             break;
                     }
+                    this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                 }
                 this.offsetStore.load();
 

diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumerTest.java
@@ -32,6 +32,7 @@
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -83,6 +84,11 @@ public void terminate() {
         pullConsumer.shutdown();
     }
 
+    @Test
+    public void testStart_OffsetShouldNotNUllAfterStart() {
+        Assert.assertNotNull(pullConsumer.getOffsetStore());
+    }
+
     @Test
     public void testPullMessage_Success() throws Exception {
         doAnswer(new Answer() {

diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
@@ -52,6 +52,7 @@
 import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
 import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -154,6 +155,11 @@ public void terminate() {
         pushConsumer.shutdown();
     }
 
+    @Test
+    public void testStart_OffsetShouldNotNUllAfterStart() {
+        Assert.assertNotNull(pushConsumer.getOffsetStore());
+    }
+
     @Test
     public void testPullMessage_Success() throws InterruptedException, RemotingException, MQBrokerException {
         final CountDownLatch countDownLatch = new CountDownLatch(1);