[Issue #386] fixing ConsumerGroup Queue Consumer Offset not synced up issue

eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/MeshAsyncConsumeContext.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.api;
+
+import io.openmessaging.api.AsyncConsumeContext;
+
+public abstract class MeshAsyncConsumeContext extends AsyncConsumeContext {
+    private AbstractContext context;
+
+    public AbstractContext getContext() {
+        return context;
+    }
+
+    public void setContext(AbstractContext context) {
+        this.context = context;
+    }
+}

eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/consumer/MeshMQPushConsumer.java

@@ -43,6 +43,4 @@ public interface MeshMQPushConsumer extends Consumer {
 
     @Override
     void unsubscribe(String topic);
-
-    AbstractContext getContext();
 }

eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/PushConsumerImpl.java

@@ -34,6 +34,7 @@
 import io.openmessaging.api.exception.OMSRuntimeException;
 
 import org.apache.eventmesh.api.AbstractContext;
+import org.apache.eventmesh.api.MeshAsyncConsumeContext;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.connector.rocketmq.common.EventMeshConstants;
 import org.apache.eventmesh.connector.rocketmq.config.ClientConfig;
@@ -55,7 +56,6 @@ public class PushConsumerImpl implements Consumer {
     private AtomicBoolean started = new AtomicBoolean(false);
     private final Map<String, AsyncMessageListener> subscribeTable = new ConcurrentHashMap<>();
     private final ClientConfig clientConfig;
-    private EventMeshConsumeConcurrentlyContext context;
 
     public PushConsumerImpl(final Properties properties) {
         this.rocketmqPushConsumer = new DefaultMQPushConsumer();
@@ -93,7 +93,6 @@ public PushConsumerImpl(final Properties properties) {
 
                 @Override
                 public EventMeshConsumeConcurrentlyStatus handleMessage(MessageExt msg, EventMeshConsumeConcurrentlyContext context) {
-                    PushConsumerImpl.this.setContext(context);
                     if (msg == null) {
                         return EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                     }
@@ -117,12 +116,13 @@ public EventMeshConsumeConcurrentlyStatus handleMessage(MessageExt msg, EventMes
 
                     final Properties contextProperties = new Properties();
                     contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.RECONSUME_LATER.name());
-                    AsyncConsumeContext omsContext = new AsyncConsumeContext() {
+                    MeshAsyncConsumeContext omsContext = new MeshAsyncConsumeContext() {
                         @Override
                         public void commit(Action action) {
                             contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS.name());
                         }
                     };
+                    omsContext.setContext(context);
                     listener.consume(omsMsg, omsContext);
 
                     return EventMeshConsumeConcurrentlyStatus.valueOf(contextProperties.getProperty(NonStandardKeys.MESSAGE_CONSUME_STATUS));
@@ -133,7 +133,6 @@ public void commit(Action action) {
 
                 @Override
                 public EventMeshConsumeConcurrentlyStatus handleMessage(MessageExt msg, EventMeshConsumeConcurrentlyContext context) {
-                    PushConsumerImpl.this.setContext(context);
                     if (msg == null) {
                         return EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                     }
@@ -157,13 +156,14 @@ public EventMeshConsumeConcurrentlyStatus handleMessage(MessageExt msg, EventMes
 
                     contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.RECONSUME_LATER.name());
 
-                    AsyncConsumeContext omsContext = new AsyncConsumeContext() {
+                    MeshAsyncConsumeContext omsContext = new MeshAsyncConsumeContext() {
                         @Override
                         public void commit(Action action) {
                             contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS,
                                     EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS.name());
                         }
                     };
+                    omsContext.setContext(context);
                     listener.consume(omsMsg, omsContext);
 
                     return EventMeshConsumeConcurrentlyStatus.valueOf(contextProperties.getProperty(NonStandardKeys.MESSAGE_CONSUME_STATUS));
@@ -254,14 +254,6 @@ public DefaultMQPushConsumer getRocketmqPushConsumer() {
 //        }
 //    }
 
-    public AbstractContext getContext() {
-        return this.context;
-    }
-
-    public void setContext(EventMeshConsumeConcurrentlyContext context) {
-        this.context = context;
-    }
-
     @Override
     public void subscribe(String topic, String subExpression, MessageListener listener) {
 

eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/RocketMQConsumerImpl.java

@@ -98,12 +98,6 @@ public void subscribe(String topic, AsyncMessageListener listener) throws Except
         pushConsumer.subscribe(topic, "*", listener);
     }
 
-    @Override
-    public AbstractContext getContext() {
-        return pushConsumer.getContext();
-    }
-
-
     @Override
     public boolean isStarted() {
         return pushConsumer.isStarted();

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQConsumerWrapper.java

@@ -88,8 +88,4 @@ public synchronized void shutdown() throws Exception {
     public void updateOffset(List<Message> msgs, AbstractContext eventMeshConsumeConcurrentlyContext) {
         meshMQPushConsumer.updateOffset(msgs, eventMeshConsumeConcurrentlyContext);
     }
-
-    public AbstractContext getContext() {
-        return meshMQPushConsumer.getContext();
-    }
 }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/EventMeshConsumer.java

@@ -31,6 +31,7 @@
 
 import org.apache.commons.collections4.MapUtils;
 import org.apache.eventmesh.api.AbstractContext;
+import org.apache.eventmesh.api.MeshAsyncConsumeContext;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
 import org.apache.eventmesh.runtime.constants.EventMeshConstants;
@@ -138,7 +139,7 @@ public void consume(Message message, AsyncConsumeContext context) {
                         }
                     }
                     HandleMsgContext handleMsgContext = new HandleMsgContext(EventMeshUtil.buildPushMsgSeqNo(), consumerGroupConf.getConsumerGroup(), EventMeshConsumer.this,
-                            topic, message, persistentMqConsumer.getContext(), consumerGroupConf, eventMeshHTTPServer, bizSeqNo, uniqueId, currentTopicConfig);
+                            topic, message, ((MeshAsyncConsumeContext)context).getContext(), consumerGroupConf, eventMeshHTTPServer, bizSeqNo, uniqueId, currentTopicConfig);
 
                     if (httpMessageHandler.handle(handleMsgContext)) {
 //                        context.attributes().put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.CONSUME_FINISH.name());
@@ -187,7 +188,7 @@ public void consume(Message message, AsyncConsumeContext context) {
                         }
                     }
                     HandleMsgContext handleMsgContext = new HandleMsgContext(EventMeshUtil.buildPushMsgSeqNo(), consumerGroupConf.getConsumerGroup(), EventMeshConsumer.this,
-                            topic, message, broadcastMqConsumer.getContext(), consumerGroupConf, eventMeshHTTPServer, bizSeqNo, uniqueId, currentTopicConfig);
+                            topic, message, ((MeshAsyncConsumeContext)context).getContext(), consumerGroupConf, eventMeshHTTPServer, bizSeqNo, uniqueId, currentTopicConfig);
 
                     if (httpMessageHandler.handle(handleMsgContext)) {
 //                        context.attributes().put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.CONSUME_FINISH.name());

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java

@@ -40,6 +40,7 @@
 
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.api.MeshAsyncConsumeContext;
 import org.apache.eventmesh.api.RRCallback;
 import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
@@ -533,7 +534,7 @@ public void consume(Message message, AsyncConsumeContext context) {
                     Iterator<Session> sessionsItr = groupConsumerSessions.iterator();
 
                     DownStreamMsgContext downStreamMsgContext =
-                            new DownStreamMsgContext(message, null, broadCastMsgConsumer, broadCastMsgConsumer.getContext(), false);
+                            new DownStreamMsgContext(message, null, broadCastMsgConsumer, ((MeshAsyncConsumeContext)context).getContext(), false);
 
                     while (sessionsItr.hasNext()) {
                         Session session = sessionsItr.next();
@@ -605,7 +606,7 @@ public void consume(Message message, AsyncConsumeContext context) {
                     }
 
                     DownStreamMsgContext downStreamMsgContext =
-                            new DownStreamMsgContext(message, session, persistentMsgConsumer, persistentMsgConsumer.getContext(), false);
+                            new DownStreamMsgContext(message, session, persistentMsgConsumer, ((MeshAsyncConsumeContext)context).getContext(), false);
                     //msg put in eventmesh,waiting client ack
                     session.getPusher().unAckMsg(downStreamMsgContext.seq, downStreamMsgContext);
                     session.downstreamMsg(downStreamMsgContext);