This is an automated email from the ASF dual-hosted git repository.
mytang0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 1780ef772 [ISSUE #3288]Refactor ConsumerGroupManager
new db5605277 Merge pull request #3289 from mxsm/eventmesh-3288
1780ef772 is described below
commit 1780ef7727e8aa5353f1f4e2bd42c0edae98c681
Author: mxsm <[email protected]>
AuthorDate: Mon Feb 27 23:27:01 2023 +0800
[ISSUE #3288]Refactor ConsumerGroupManager
---
.../http/consumer/ConsumerGroupManager.java | 33 +++++++++++++---------
1 file changed, 20 insertions(+), 13 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerGroupManager.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerGroupManager.java
index 640ed86ac..cca28c72f 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerGroupManager.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerGroupManager.java
@@ -29,31 +29,36 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class ConsumerGroupManager {
- protected AtomicBoolean started = new AtomicBoolean(Boolean.FALSE);
+ private final AtomicBoolean started = new AtomicBoolean(false);
- protected AtomicBoolean inited = new AtomicBoolean(Boolean.FALSE);
+ private final AtomicBoolean inited = new AtomicBoolean(false);
- private EventMeshHTTPServer eventMeshHTTPServer;
+ private final EventMeshHTTPServer eventMeshHTTPServer;
- private EventMeshConsumer eventMeshConsumer;
+ private final EventMeshConsumer eventMeshConsumer;
private ConsumerGroupConf consumerGroupConfig;
- public ConsumerGroupManager(EventMeshHTTPServer eventMeshHTTPServer,
ConsumerGroupConf consumerGroupConfig) {
+ public ConsumerGroupManager(final EventMeshHTTPServer eventMeshHTTPServer,
final ConsumerGroupConf consumerGroupConfig) {
this.eventMeshHTTPServer = eventMeshHTTPServer;
this.consumerGroupConfig = consumerGroupConfig;
- eventMeshConsumer = new EventMeshConsumer(this.eventMeshHTTPServer,
this.consumerGroupConfig);
+ this.eventMeshConsumer = new
EventMeshConsumer(this.eventMeshHTTPServer, this.consumerGroupConfig);
}
- public synchronized void init() throws Exception {
+ public void init() throws Exception {
+ if (!inited.compareAndSet(false, true)) {
+ return;
+ }
eventMeshConsumer.init();
- inited.compareAndSet(false, true);
+
}
- public synchronized void start() throws Exception {
+ public void start() throws Exception {
+ if (!started.compareAndSet(false, true)) {
+ return;
+ }
setupEventMeshConsumer(consumerGroupConfig);
eventMeshConsumer.start();
- started.compareAndSet(false, true);
}
private synchronized void setupEventMeshConsumer(ConsumerGroupConf
consumerGroupConfig) throws Exception {
@@ -62,12 +67,14 @@ public class ConsumerGroupManager {
}
}
- public synchronized void shutdown() throws Exception {
+ public void shutdown() throws Exception {
+ if (!started.compareAndSet(true, false)) {
+ return;
+ }
eventMeshConsumer.shutdown();
- started.compareAndSet(true, false);
}
- public synchronized void refresh(ConsumerGroupConf consumerGroupConfig)
throws Exception {
+ public synchronized void refresh(final ConsumerGroupConf
consumerGroupConfig) throws Exception {
if (consumerGroupConfig == null ||
this.consumerGroupConfig.equals(consumerGroupConfig)) {
return;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
