fix(example) fix example: RocketMQSourceExample1 (#100)

Co-authored-by: 维章 <[email protected]>

Author: ni-ze

rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/source/RocketMQSource.java

@@ -287,13 +287,17 @@ protected void setOffsetStore(DefaultMQPushConsumer consumer) {
         }
         MQClientInstance mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(defaultMQPushConsumer.getDefaultMQPushConsumer());
         RemoteBrokerOffsetStore offsetStore = new RemoteBrokerOffsetStore(mQClientFactory, NamespaceUtil.wrapNamespace(consumer.getNamespace(), consumer.getConsumerGroup())) {
-
+            Set<MessageQueue> firstComing = new HashSet<>();
             @Override
             public void removeOffset(MessageQueue mq) {
-                Set<String> splitIds = new HashSet<>();
-                splitIds.add(new RocketMQMessageQueue(mq).getQueueId());
-                removeSplit(splitIds);
-                super.removeOffset(mq);
+                if (!firstComing.contains(mq)){
+                    firstComing.add(mq);
+                } else {
+                    Set<String> splitIds = new HashSet<>();
+                    splitIds.add(new RocketMQMessageQueue(mq).getQueueId());
+                    removeSplit(splitIds);
+                    super.removeOffset(mq);
+                }
             }
 
             @Override
@@ -398,4 +402,16 @@ public String getConsumerOffset() {
     public void setConsumerOffset(String consumerOffset) {
         this.consumerOffset = consumerOffset;
     }
+
+    public String getStrategyName() {
+        return strategyName;
+    }
+
+    public void setStrategyName(String strategyName) {
+        this.strategyName = strategyName;
+    }
+
+    public void setConsumer(DefaultMQPushConsumer consumer) {
+        this.consumer = consumer;
+    }
 }