Consumer synchronized start or stop

openmessaging/src/main/java/io/openmessaging/rocketmq/config/DefaultQueueMetaData.java

@@ -38,7 +38,7 @@ public class DefaultQueueMetaData implements QueueMetaData {
 
     @Override
     public List<QueueMetaData.Partition> partitions() {
-        return null;
+        return partitions;
     }
 
     public static class DefaultPartition implements Partition {

openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PullConsumerImpl.java

@@ -93,7 +93,7 @@ public class PullConsumerImpl implements Consumer {
 
         String consumerId = OMSUtil.buildInstanceName();
         this.rocketmqPullConsumer.setInstanceName(consumerId);
-        properties.put("TIMEOUT", consumerId);
+        properties.put("CONSUMER_ID", consumerId);
 
         this.rocketmqPullConsumer.setLanguage(LanguageCode.OMS);
 
@@ -325,7 +325,7 @@ public class PullConsumerImpl implements Consumer {
     }
 
     @Override
-    public void start() {
+    public synchronized void start() {
         if (!started) {
             try {
                 this.pullConsumerScheduleService.start();
@@ -339,7 +339,7 @@ public class PullConsumerImpl implements Consumer {
     }
 
     @Override
-    public void stop() {
+    public synchronized void stop() {
         if (this.started) {
             this.localMessageCache.stop();
             this.pullConsumerScheduleService.shutdown();

openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java

@@ -255,7 +255,7 @@ public class PushConsumerImpl implements Consumer {
     }
 
     @Override
-    public void start() {
+    public synchronized void start() {
         currentState = ServiceLifeState.STARTING;
         if (!started) {
             try {
@@ -269,7 +269,7 @@ public class PushConsumerImpl implements Consumer {
     }
 
     @Override
-    public void stop() {
+    public synchronized void stop() {
         currentState = ServiceLifeState.STOPPING;
         if (this.started) {
             this.rocketmqPushConsumer.shutdown();