You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2021/06/16 07:44:59 UTC
[incubator-eventmesh] branch develop updated: [Issue #386] fixing
ConsumerGroup Queue Consumer Offset not synced up issue (#387)
This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/develop by this push:
new 38643e6 [Issue #386] fixing ConsumerGroup Queue Consumer Offset not synced up issue (#387)
38643e6 is described below
commit 38643e69740b9184f4b2162919a9db442ca5c4e7
Author: jinrongluo <ka...@gmail.com>
AuthorDate: Wed Jun 16 03:43:44 2021 -0400
[Issue #386] fixing ConsumerGroup Queue Consumer Offset not synced up issue (#387)
* [Issue #337] Fix HttpSubscriber startup issue
* [Issue #337] test commit
* [Issue #337] revert test commit
* [Issue #337] Enhance Http Demo Subscriber by using ExecutorService, CountDownLatch and PreDestroy hook
* [Issue #337] Enhance Http Demo Subscriber by using ExecutorService, CountDownLatch and PreDestroy hook
* [Issue #337] Address code review comment for Subscriber Demo App
* [Issue #386] fixing ConsumerGroup Queuen Consumer Offset not synced up issue
* [Issue #386] adding license header to new file
* [Issue #386] Fix license header missing issue
Co-authored-by: j00441484 <ji...@huawei.com>
---
.../eventmesh/api/MeshAsyncConsumeContext.java | 32 ++++++++++++++++++++++
.../eventmesh/api/consumer/MeshMQPushConsumer.java | 2 --
.../rocketmq/consumer/PushConsumerImpl.java | 18 ++++--------
.../rocketmq/consumer/RocketMQConsumerImpl.java | 6 ----
.../runtime/core/plugin/MQConsumerWrapper.java | 4 ---
.../protocol/http/consumer/EventMeshConsumer.java | 5 ++--
.../tcp/client/group/ClientGroupWrapper.java | 5 ++--
7 files changed, 43 insertions(+), 29 deletions(-)
diff --git a/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/MeshAsyncConsumeContext.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/MeshAsyncConsumeContext.java
new file mode 100644
index 0000000..d6b2aa4
--- /dev/null
+++ b/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;
+ }
+}
diff --git a/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/consumer/MeshMQPushConsumer.java b/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/consumer/MeshMQPushConsumer.java
index b495169..172a674 100644
--- a/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/consumer/MeshMQPushConsumer.java
+++ b/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();
}
diff --git a/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/PushConsumerImpl.java b/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/PushConsumerImpl.java
index 2193ca1..bbdbd86 100644
--- a/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/PushConsumerImpl.java
+++ b/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/PushConsumerImpl.java
@@ -34,6 +34,7 @@ import io.openmessaging.api.MessageSelector;
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 class PushConsumerImpl implements Consumer {
@Override
public EventMeshConsumeConcurrentlyStatus handleMessage(MessageExt msg, EventMeshConsumeConcurrentlyContext context) {
- PushConsumerImpl.this.setContext(context);
if (msg == null) {
return EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
@@ -117,12 +116,13 @@ public class PushConsumerImpl implements Consumer {
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 class PushConsumerImpl implements Consumer {
@Override
public EventMeshConsumeConcurrentlyStatus handleMessage(MessageExt msg, EventMeshConsumeConcurrentlyContext context) {
- PushConsumerImpl.this.setContext(context);
if (msg == null) {
return EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
@@ -157,13 +156,14 @@ public class PushConsumerImpl implements Consumer {
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 class PushConsumerImpl implements Consumer {
// }
// }
- public AbstractContext getContext() {
- return this.context;
- }
-
- public void setContext(EventMeshConsumeConcurrentlyContext context) {
- this.context = context;
- }
-
@Override
public void subscribe(String topic, String subExpression, MessageListener listener) {
diff --git a/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/RocketMQConsumerImpl.java b/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/RocketMQConsumerImpl.java
index ce076f8..9eb6379 100644
--- a/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/RocketMQConsumerImpl.java
+++ b/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/RocketMQConsumerImpl.java
@@ -99,12 +99,6 @@ public class RocketMQConsumerImpl implements MeshMQPushConsumer {
}
@Override
- public AbstractContext getContext() {
- return pushConsumer.getContext();
- }
-
-
- @Override
public boolean isStarted() {
return pushConsumer.isStarted();
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQConsumerWrapper.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQConsumerWrapper.java
index ad44d3e..39b2c63 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQConsumerWrapper.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQConsumerWrapper.java
@@ -88,8 +88,4 @@ public class MQConsumerWrapper extends MQWrapper {
public void updateOffset(List<Message> msgs, AbstractContext eventMeshConsumeConcurrentlyContext) {
meshMQPushConsumer.updateOffset(msgs, eventMeshConsumeConcurrentlyContext);
}
-
- public AbstractContext getContext() {
- return meshMQPushConsumer.getContext();
- }
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/EventMeshConsumer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/EventMeshConsumer.java
index 186ca35..36dee3c 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/EventMeshConsumer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/EventMeshConsumer.java
@@ -31,6 +31,7 @@ import io.openmessaging.api.SendResult;
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 class EventMeshConsumer {
}
}
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 class EventMeshConsumer {
}
}
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());
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
index 07a58ac..0bc645e 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java
@@ -40,6 +40,7 @@ import io.openmessaging.api.SendResult;
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 class ClientGroupWrapper {
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 class ClientGroupWrapper {
}
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);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org