This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 4d561b814 [ISSUE #4431]Support update subscription data for
http/tcp/grpc protocol to meta storage (#4432)
4d561b814 is described below
commit 4d561b814078fa58e3479e8510292104df234b30
Author: mike_xwm <[email protected]>
AuthorDate: Thu Sep 7 10:06:10 2023 +0800
[ISSUE #4431]Support update subscription data for http/tcp/grpc protocol to
meta storage (#4432)
* update meta service
* [ISSUE #4373] Support nacos as meta storage
* [ISSUE #4372] Support etcd as Meta storage
* [ISSUE #4431] Support update subscription data for http/tcp/grpc protocol
to meta storage
* fix checkstyle error
* fix checkstyle error
* fix checkstyle error
---
.../common/config/CommonConfiguration.java | 39 +++++----
.../src/test/resources/configuration.properties | 6 +-
.../org/apache/eventmesh/api/meta/MetaService.java | 6 ++
.../api/meta/config/EventMeshMetaConfig.java | 24 ++++++
.../meta/consul/service/ConsulMetaService.java | 21 ++++-
.../src/test/java/ConsulMetaServiceTest.java | 2 +-
.../meta/etcd/service/EtcdMetaService.java | 30 ++++++-
.../registry/etcd/service/EtcdMetaServiceTest.java | 2 +-
...ion.java => NacosMetaStorageConfiguration.java} | 4 +-
.../meta/nacos/service/NacosMetaService.java | 93 +++++++++++++++++-----
.../nacos/service/NacosMetaServiceTest.java | 4 +-
.../zookeeper/service/ZookeeperMetaService.java | 21 ++++-
.../service/ZookeeperMetaServiceTest.java | 2 +-
.../admin/handler/ConfigurationHandler.java | 2 +-
.../runtime/boot/EventMeshGrpcServer.java | 2 +-
.../runtime/boot/EventMeshHTTPServer.java | 4 +-
.../eventmesh/runtime/boot/EventMeshTCPServer.java | 2 +-
.../runtime/core/consumer/SubscriptionManager.java | 55 +++++++++++++
.../protocol/grpc/consumer/EventMeshConsumer.java | 65 ++++++++++++++-
.../consumer/consumergroup/StreamTopicConfig.java | 4 +
.../processor/LocalSubscribeEventProcessor.java | 2 +-
.../processor/LocalUnSubscribeEventProcessor.java | 2 +-
.../http/processor/SubscribeProcessor.java | 1 +
.../http/processor/UnSubscribeProcessor.java | 1 +
.../http/processor/inf/AbstractEventProcessor.java | 50 +-----------
.../tcp/client/group/ClientGroupWrapper.java | 3 +-
.../client/group/ClientSessionGroupMapping.java | 50 ++++++++++++
.../tcp/client/processor/HelloProcessor.java | 2 +-
.../tcp/client/processor/SubscribeProcessor.java | 4 +-
.../client/rebalance/EventMeshRebalanceImpl.java | 4 +-
.../client/recommend/EventMeshRecommendImpl.java | 4 +-
.../apache/eventmesh/runtime/meta/MetaStorage.java | 4 +
.../src/test/resources/configuration.properties | 13 +--
33 files changed, 397 insertions(+), 131 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
index 664fd2121..350e8c733 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
@@ -53,9 +53,26 @@ public class CommonConfiguration {
@ConfigFiled(field = "server.hostIp", reload = true)
private String eventMeshServerIp = null;
- @ConfigFiled(field = "registry.plugin.server-addr", notEmpty = true)
- private String namesrvAddr = "";
+ @ConfigFiled(field = "metaStorage.plugin.server-addr", notEmpty = true)
+ private String metaStorageAddr = "";
+ @ConfigFiled(field = "metaStorage.plugin.type", notEmpty = true)
+ private String eventMeshMetaStoragePluginType = "namesrv";
+
+ @ConfigFiled(field = "metaStorage.plugin.username")
+ private String eventMeshMetaStoragePluginUsername = "";
+
+ @ConfigFiled(field = "metaStorage.plugin.password")
+ private String eventMeshMetaStoragePluginPassword = "";
+
+ @ConfigFiled(field = "metaStorage.plugin.metaStorageIntervalInMills")
+ private Integer eventMeshMetaStorageIntervalInMills = 10 * 1000;
+
+ @ConfigFiled(field =
"metaStorage.plugin.fetchMetaStorageAddrIntervalInMills")
+ private Integer eventMeshFetchMetaStorageAddrInterval = 10 * 1000;
+
+ @ConfigFiled(field = "metaStorage.plugin.enabled")
+ private boolean eventMeshServerMetaStorageEnable = false;
@ConfigFiled(field = "trace.plugin", notEmpty = true)
private String eventMeshTracePluginType;
@@ -63,9 +80,6 @@ public class CommonConfiguration {
@ConfigFiled(field = "metrics.plugin", notEmpty = true)
private List<String> eventMeshMetricsPluginType;
- @ConfigFiled(field = "metaStorage.plugin.type", notEmpty = true)
- private String eventMeshMetaStoragePluginType = "namesrv";
-
@ConfigFiled(field = "security.plugin.type", notEmpty = true)
private String eventMeshSecurityPluginType = "security";
@@ -78,18 +92,6 @@ public class CommonConfiguration {
@ConfigFiled(field = "security.validation.type.token", notEmpty = true)
private boolean eventMeshSecurityValidateTypeToken = false;
- @ConfigFiled(field = "metaStorage.plugin.username")
- private String eventMeshMetaStoragePluginUsername = "";
-
- @ConfigFiled(field = "metaStorage.plugin.password")
- private String eventMeshMetaStoragePluginPassword = "";
-
- @ConfigFiled(field = "server.metaStorage.metaStorageIntervalInMills")
- private Integer eventMeshMetaStorageIntervalInMills = 10 * 1000;
-
- @ConfigFiled(field =
"server.metaStorage.fetchMetaStorageAddrIntervalInMills")
- private Integer eventMeshFetchMetaStorageAddrInterval = 10 * 1000;
-
@ConfigFiled(field = "server.trace.enabled")
private boolean eventMeshServerTraceEnable = false;
@@ -97,9 +99,6 @@ public class CommonConfiguration {
@ConfigFiled(field = "server.security.enabled")
private boolean eventMeshServerSecurityEnable = false;
- @ConfigFiled(field = "server.metaStorage.enabled")
- private boolean eventMeshServerMetaStorageEnable = false;
-
@ConfigFiled(field = "security.publickey")
private String eventMeshSecurityPublickey = "";
diff --git a/eventmesh-common/src/test/resources/configuration.properties
b/eventmesh-common/src/test/resources/configuration.properties
index 7266c8604..f80bb7db0 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-common/src/test/resources/configuration.properties
@@ -25,13 +25,13 @@ eventMesh.storage.plugin.type=storage-succeed!!!
eventMesh.security.plugin.type=security-succeed!!!
eventMesh.metaStorage.plugin.type=metaStorage-succeed!!!
eventMesh.trace.plugin=trace-succeed!!!
-eventMesh.server.metaStorage.metaStorageIntervalInMills=816
-eventMesh.server.metaStorage.fetchMetaStorageAddrIntervalInMills=1816
+eventMesh.metaStorage.plugin.metaStorageIntervalInMills=816
+eventMesh.metaStorage.plugin.fetchMetaStorageAddrIntervalInMills=1816
eventMesh.metrics.plugin=metrics-succeed1!!!,metrics-succeed2!!!,metrics-succeed3!!!
eventMesh.metaStorage.plugin.server-addr=server-addr-succeed1!!!
eventMesh.server.security.enabled=true
-eventMesh.server.metaStorage.enabled=true
+eventMesh.metaStorage.plugin.enabled=true
eventMesh.server.trace.enabled=true
eventMesh.server.provide.protocols=TCP,HTTP,GRPC
diff --git
a/eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/MetaService.java
b/eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/MetaService.java
index a659e6205..66c8a203f 100644
---
a/eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/MetaService.java
+++
b/eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/MetaService.java
@@ -54,6 +54,12 @@ public interface MetaService {
void registerMetadata(Map<String, String> metadataMap);
+ String getMetaData(String key);
+
+ void updateMetaData(Map<String, String> metadataMap);
+
+ void removeMetaData(String key);
+
boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws
MetaException;
boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throws
MetaException;
diff --git
a/eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/config/EventMeshMetaConfig.java
b/eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/config/EventMeshMetaConfig.java
new file mode 100644
index 000000000..d8ea8c8f8
--- /dev/null
+++
b/eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/config/EventMeshMetaConfig.java
@@ -0,0 +1,24 @@
+/*
+ * 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.meta.config;
+
+public class EventMeshMetaConfig {
+
+ public static final String EVENT_MESH_PROTO = "protocol";
+
+}
diff --git
a/eventmesh-meta/eventmesh-meta-consul/src/main/java/org/apache/eventmesh/meta/consul/service/ConsulMetaService.java
b/eventmesh-meta/eventmesh-meta-consul/src/main/java/org/apache/eventmesh/meta/consul/service/ConsulMetaService.java
index 2e3dbba18..b2ba4564d 100644
---
a/eventmesh-meta/eventmesh-meta-consul/src/main/java/org/apache/eventmesh/meta/consul/service/ConsulMetaService.java
+++
b/eventmesh-meta/eventmesh-meta-consul/src/main/java/org/apache/eventmesh/meta/consul/service/ConsulMetaService.java
@@ -72,11 +72,11 @@ public class ConsulMetaService implements MetaService {
for (String key : ConfigurationContextUtil.KEYS) {
CommonConfiguration commonConfiguration =
ConfigurationContextUtil.get(key);
if (null != commonConfiguration) {
- String namesrvAddr = commonConfiguration.getNamesrvAddr();
- if (StringUtils.isBlank(namesrvAddr)) {
+ String metaStorageAddr =
commonConfiguration.getMetaStorageAddr();
+ if (StringUtils.isBlank(metaStorageAddr)) {
throw new MetaException("namesrvAddr cannot be null");
}
- String[] addr = namesrvAddr.split(":");
+ String[] addr = metaStorageAddr.split(":");
if (addr.length != 2) {
throw new MetaException("Illegal namesrvAddr");
}
@@ -190,4 +190,19 @@ public class ConsulMetaService implements MetaService {
public void registerMetadata(Map<String, String> metadataMap) {
}
+
+ @Override
+ public String getMetaData(String key) {
+ return null;
+ }
+
+ @Override
+ public void updateMetaData(Map<String, String> metadataMap) {
+
+ }
+
+ @Override
+ public void removeMetaData(String key) {
+
+ }
}
diff --git
a/eventmesh-meta/eventmesh-meta-consul/src/test/java/ConsulMetaServiceTest.java
b/eventmesh-meta/eventmesh-meta-consul/src/test/java/ConsulMetaServiceTest.java
index 9bfbe6631..bb34fdd4b 100644
---
a/eventmesh-meta/eventmesh-meta-consul/src/test/java/ConsulMetaServiceTest.java
+++
b/eventmesh-meta/eventmesh-meta-consul/src/test/java/ConsulMetaServiceTest.java
@@ -50,7 +50,7 @@ public class ConsulMetaServiceTest {
consulMetaService = new ConsulMetaService();
CommonConfiguration configuration = new CommonConfiguration();
ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.HTTP,
configuration);
- configuration.setNamesrvAddr("127.0.0.1:8500");
+ configuration.setMetaStorageAddr("127.0.0.1:8500");
Mockito.when(eventMeshRegisterInfo.getEventMeshClusterName()).thenReturn("eventmesh");
Mockito.when(eventMeshRegisterInfo.getEventMeshName()).thenReturn("eventmesh");
Mockito.when(eventMeshRegisterInfo.getEndPoint()).thenReturn("127.0.0.1:8500");
diff --git
a/eventmesh-meta/eventmesh-meta-etcd/src/main/java/org/apache/eventmesh/meta/etcd/service/EtcdMetaService.java
b/eventmesh-meta/eventmesh-meta-etcd/src/main/java/org/apache/eventmesh/meta/etcd/service/EtcdMetaService.java
index 8e1d9af97..fae37e27c 100644
---
a/eventmesh-meta/eventmesh-meta-etcd/src/main/java/org/apache/eventmesh/meta/etcd/service/EtcdMetaService.java
+++
b/eventmesh-meta/eventmesh-meta-etcd/src/main/java/org/apache/eventmesh/meta/etcd/service/EtcdMetaService.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.ThreadPoolFactory;
import org.apache.eventmesh.common.config.CommonConfiguration;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+import org.apache.eventmesh.common.utils.IPUtils;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.meta.etcd.constant.EtcdConstant;
import org.apache.eventmesh.meta.etcd.factory.EtcdClientFactory;
@@ -33,6 +34,7 @@ import
org.apache.eventmesh.meta.etcd.factory.EtcdClientFactory;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -70,6 +72,10 @@ public class EtcdMetaService implements MetaService {
private String password;
+ private String instanceIp;
+
+ private String group;
+
@Getter
private Client etcdClient;
@@ -89,12 +95,14 @@ public class EtcdMetaService implements MetaService {
if (null == commonConfiguration) {
continue;
}
- if (StringUtils.isBlank(commonConfiguration.getNamesrvAddr())) {
+ if (StringUtils.isBlank(commonConfiguration.getMetaStorageAddr()))
{
throw new MetaException("namesrvAddr cannot be null");
}
- this.serverAddr = commonConfiguration.getNamesrvAddr();
+ this.serverAddr = commonConfiguration.getMetaStorageAddr();
this.username =
commonConfiguration.getEventMeshMetaStoragePluginUsername();
this.password =
commonConfiguration.getEventMeshMetaStoragePluginPassword();
+ this.instanceIp = IPUtils.getLocalAddress();
+ this.group = commonConfiguration.getMeshGroup();
break;
}
etcdRegistryMonitorExecutorService =
ThreadPoolFactory.createSingleScheduledExecutor(
@@ -188,6 +196,24 @@ public class EtcdMetaService implements MetaService {
}
}
+ @Override
+ public String getMetaData(String key) {
+ return null;
+ }
+
+ @Override
+ public void updateMetaData(Map<String, String> metadataMap) {
+ String etcdMetaKey = instanceIp + "-" + group;
+ ByteSequence key = ByteSequence.from(etcdMetaKey,
StandardCharsets.UTF_8);
+ ByteSequence value =
ByteSequence.from(Objects.requireNonNull(JsonUtils.toJSONString(metadataMap)),
StandardCharsets.UTF_8);
+ etcdClient.getKVClient().put(key, value);
+ }
+
+ @Override
+ public void removeMetaData(String key) {
+
+ }
+
@Override
public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo)
throws MetaException {
String eventMeshClusterName =
eventMeshRegisterInfo.getEventMeshClusterName();
diff --git
a/eventmesh-meta/eventmesh-meta-etcd/src/test/java/org/apache/eventmesh/registry/etcd/service/EtcdMetaServiceTest.java
b/eventmesh-meta/eventmesh-meta-etcd/src/test/java/org/apache/eventmesh/registry/etcd/service/EtcdMetaServiceTest.java
index 7a7da117a..9eef2c8b8 100644
---
a/eventmesh-meta/eventmesh-meta-etcd/src/test/java/org/apache/eventmesh/registry/etcd/service/EtcdMetaServiceTest.java
+++
b/eventmesh-meta/eventmesh-meta-etcd/src/test/java/org/apache/eventmesh/registry/etcd/service/EtcdMetaServiceTest.java
@@ -50,7 +50,7 @@ public class EtcdMetaServiceTest {
public void setUp() {
etcdMetaService = new EtcdMetaService();
CommonConfiguration configuration = new CommonConfiguration();
- configuration.setNamesrvAddr("127.0.0.1:2379");
+ configuration.setMetaStorageAddr("127.0.0.1:2379");
ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.HTTP,
configuration);
//
Mockito.when(eventMeshRegisterInfo.getEventMeshClusterName()).thenReturn("eventmesh");
diff --git
a/eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/config/NacosRegistryConfiguration.java
b/eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/config/NacosMetaStorageConfiguration.java
similarity index 95%
rename from
eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/config/NacosRegistryConfiguration.java
rename to
eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/config/NacosMetaStorageConfiguration.java
index 0f72ede9b..67e51d2d9 100644
---
a/eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/config/NacosRegistryConfiguration.java
+++
b/eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/config/NacosMetaStorageConfiguration.java
@@ -28,8 +28,8 @@ import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
-@Config(prefix = "eventMesh.registry.nacos")
-public class NacosRegistryConfiguration {
+@Config(prefix = "eventMesh.metaStorage.nacos")
+public class NacosMetaStorageConfiguration {
@ConfigFiled(field = PropertyKeyConst.ENDPOINT)
private String endpoint;
diff --git
a/eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/service/NacosMetaService.java
b/eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/service/NacosMetaService.java
index b64d5ccca..8d0a61fc9 100644
---
a/eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/service/NacosMetaService.java
+++
b/eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/service/NacosMetaService.java
@@ -19,13 +19,15 @@ package org.apache.eventmesh.meta.nacos.service;
import org.apache.eventmesh.api.exception.MetaException;
import org.apache.eventmesh.api.meta.MetaService;
+import org.apache.eventmesh.api.meta.config.EventMeshMetaConfig;
import org.apache.eventmesh.api.meta.dto.EventMeshDataInfo;
import org.apache.eventmesh.api.meta.dto.EventMeshRegisterInfo;
import org.apache.eventmesh.api.meta.dto.EventMeshUnRegisterInfo;
import org.apache.eventmesh.common.config.CommonConfiguration;
import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
-import org.apache.eventmesh.meta.nacos.config.NacosRegistryConfiguration;
+import org.apache.eventmesh.common.utils.IPUtils;
+import org.apache.eventmesh.meta.nacos.config.NacosMetaStorageConfiguration;
import org.apache.eventmesh.meta.nacos.constant.NacosConstant;
import org.apache.commons.lang3.StringUtils;
@@ -48,6 +50,7 @@ import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.client.naming.utils.UtilAndComs;
import com.alibaba.nacos.common.utils.CollectionUtils;
+import com.alibaba.nacos.common.utils.JacksonUtils;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -55,9 +58,9 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class NacosMetaService implements MetaService {
- private final AtomicBoolean initStatus = new AtomicBoolean(false);
+ private final AtomicBoolean initStatus = new AtomicBoolean(false);
- private final AtomicBoolean startStatus = new AtomicBoolean(false);
+ private final AtomicBoolean startStatus = new AtomicBoolean(false);
@Getter
private String serverAddr;
@@ -69,10 +72,16 @@ public class NacosMetaService implements MetaService {
private String password;
@Getter
- private NacosRegistryConfiguration nacosConfig;
+ private NacosMetaStorageConfiguration nacosConfig;
+
+ private String dataId;
+
+ private String group;
@Getter
- private NamingService namingService;
+ private NamingService nacosNamingService;
+
+ private com.alibaba.nacos.api.config.ConfigService nacosConfigService;
private ConcurrentMap<String, EventMeshRegisterInfo>
eventMeshRegisterInfoMap;
@@ -88,17 +97,19 @@ public class NacosMetaService implements MetaService {
if (null == commonConfiguration) {
continue;
}
- if (StringUtils.isBlank(commonConfiguration.getNamesrvAddr())) {
+ if (StringUtils.isBlank(commonConfiguration.getMetaStorageAddr()))
{
throw new MetaException("namesrvAddr cannot be null");
}
- this.serverAddr = commonConfiguration.getNamesrvAddr();
+ this.serverAddr = commonConfiguration.getMetaStorageAddr();
this.username =
commonConfiguration.getEventMeshMetaStoragePluginUsername();
this.password =
commonConfiguration.getEventMeshMetaStoragePluginPassword();
+ this.dataId = IPUtils.getLocalAddress();
+ this.group = commonConfiguration.getMeshGroup();
break;
}
ConfigService configService = ConfigService.getInstance();
- NacosRegistryConfiguration nacosConfig =
configService.buildConfigInstance(NacosRegistryConfiguration.class);
+ NacosMetaStorageConfiguration nacosConfig =
configService.buildConfigInstance(NacosMetaStorageConfiguration.class);
if (nacosConfig != null) {
this.nacosConfig = nacosConfig;
}
@@ -110,13 +121,21 @@ public class NacosMetaService implements MetaService {
if (!startStatus.compareAndSet(false, true)) {
return;
}
+ Properties properties = buildProperties();
+ // registry
try {
- Properties properties = buildProperties();
- namingService = NacosFactory.createNamingService(properties);
+ this.nacosNamingService =
NacosFactory.createNamingService(properties);
} catch (NacosException e) {
log.error("[NacosRegistryService][start] error", e);
throw new MetaException(e.getMessage());
}
+ // config
+ try {
+ this.nacosConfigService =
NacosFactory.createConfigService(properties);
+ } catch (NacosException e) {
+ log.error("[NacosConfigService][start] error", e);
+ throw new MetaException(e.getMessage());
+ }
}
private Properties buildProperties() {
@@ -163,7 +182,7 @@ public class NacosMetaService implements MetaService {
return;
}
try {
- namingService.shutDown();
+ nacosNamingService.shutDown();
} catch (NacosException e) {
log.error("[NacosRegistryService][shutdown] error", e);
throw new MetaException(e.getMessage());
@@ -191,18 +210,18 @@ public class NacosMetaService implements MetaService {
String eventMeshName = configuration.getEventMeshName();
try {
List<Instance> instances =
- namingService.selectInstances(eventMeshName + "-" +
key,
- key + "-" + (inCluster ?
configuration.getEventMeshCluster() : NacosConstant.GROUP),
- clusters,
- true);
+ nacosNamingService.selectInstances(eventMeshName + "-" +
key,
+ key + "-" + (inCluster ?
configuration.getEventMeshCluster() : NacosConstant.GROUP),
+ clusters,
+ true);
if (CollectionUtils.isEmpty(instances)) {
continue;
}
for (Instance instance : instances) {
EventMeshDataInfo eventMeshDataInfo =
- new EventMeshDataInfo(instance.getClusterName(),
instance.getServiceName(),
- instance.getIp() + ":"
- + instance.getPort(), 0L,
instance.getMetadata());
+ new EventMeshDataInfo(instance.getClusterName(),
instance.getServiceName(),
+ instance.getIp() + ":"
+ + instance.getPort(), 0L,
instance.getMetadata());
eventMeshDataInfoList.add(eventMeshDataInfo);
}
} catch (NacosException e) {
@@ -222,11 +241,43 @@ public class NacosMetaService implements MetaService {
}
}
+ @Override
+ public String getMetaData(String key) {
+ try {
+ return this.nacosConfigService.getConfig(key, group, 5000L);
+ } catch (NacosException e) {
+ log.error("get metaData fail", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void updateMetaData(Map<String, String> metadataMap) {
+ String protocol =
metadataMap.get(EventMeshMetaConfig.EVENT_MESH_PROTO);
+ String nacosDataId = dataId + "-" + protocol;
+ try {
+ boolean flag = this.nacosConfigService.publishConfig(nacosDataId,
group, JacksonUtils.toJson(metadataMap));
+ if (flag) {
+ log.info("publish metaData {} success", metadataMap);
+ } else {
+ log.error("publish metaData {} fail", metadataMap);
+ }
+ } catch (NacosException e) {
+ log.error("failed to publish data to nacos", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void removeMetaData(String key) {
+
+ }
+
@Override
public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo)
throws MetaException {
try {
String[] ipPort =
eventMeshRegisterInfo.getEndPoint().split(NacosConstant.IP_PORT_SEPARATOR);
- if (ipPort == null || ipPort.length < 2) {
+ if (ipPort.length < 2) {
return false;
}
String eventMeshClusterName =
eventMeshRegisterInfo.getEventMeshClusterName();
@@ -240,7 +291,7 @@ public class NacosMetaService implements MetaService {
instance.setMetadata(metadata);
String eventMeshName = eventMeshRegisterInfo.getEventMeshName();
- namingService.registerInstance(eventMeshName,
eventMeshRegisterInfo.getProtocolType() + "-"
+ nacosNamingService.registerInstance(eventMeshName,
eventMeshRegisterInfo.getProtocolType() + "-"
+ NacosConstant.GROUP, instance);
eventMeshRegisterInfoMap.put(eventMeshName, eventMeshRegisterInfo);
} catch (NacosException e) {
@@ -261,7 +312,7 @@ public class NacosMetaService implements MetaService {
String eventMeshName = eventMeshUnRegisterInfo.getEventMeshName();
String eventMeshClusterName =
eventMeshUnRegisterInfo.getEventMeshClusterName();
instance.setClusterName(eventMeshClusterName);
- namingService.deregisterInstance(eventMeshName,
eventMeshUnRegisterInfo.getProtocolType()
+ nacosNamingService.deregisterInstance(eventMeshName,
eventMeshUnRegisterInfo.getProtocolType()
+ "-" + NacosConstant.GROUP, instance);
eventMeshRegisterInfoMap.remove(eventMeshName);
} catch (NacosException e) {
diff --git
a/eventmesh-meta/eventmesh-meta-nacos/src/test/java/org/apache/eventmesh/registry/nacos/service/NacosMetaServiceTest.java
b/eventmesh-meta/eventmesh-meta-nacos/src/test/java/org/apache/eventmesh/registry/nacos/service/NacosMetaServiceTest.java
index 778ffaccb..7ec238da2 100644
---
a/eventmesh-meta/eventmesh-meta-nacos/src/test/java/org/apache/eventmesh/registry/nacos/service/NacosMetaServiceTest.java
+++
b/eventmesh-meta/eventmesh-meta-nacos/src/test/java/org/apache/eventmesh/registry/nacos/service/NacosMetaServiceTest.java
@@ -49,7 +49,7 @@ public class NacosMetaServiceTest {
public void setUp() {
nacosMetaService = new NacosMetaService();
CommonConfiguration configuration = new CommonConfiguration();
- configuration.setNamesrvAddr("127.0.0.1");
+ configuration.setMetaStorageAddr("127.0.0.1");
configuration.setEventMeshMetaStoragePluginPassword("nacos");
configuration.setEventMeshMetaStoragePluginUsername("nacos");
ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.HTTP,
configuration);
@@ -80,7 +80,7 @@ public class NacosMetaServiceTest {
public void testStart() {
nacosMetaService.init();
nacosMetaService.start();
- Assert.assertNotNull(nacosMetaService.getNamingService());
+ Assert.assertNotNull(nacosMetaService.getNacosNamingService());
}
diff --git
a/eventmesh-meta/eventmesh-meta-zookeeper/src/main/java/org/apache/eventmesh/meta/zookeeper/service/ZookeeperMetaService.java
b/eventmesh-meta/eventmesh-meta-zookeeper/src/main/java/org/apache/eventmesh/meta/zookeeper/service/ZookeeperMetaService.java
index 701a6a918..a5f57f10f 100644
---
a/eventmesh-meta/eventmesh-meta-zookeeper/src/main/java/org/apache/eventmesh/meta/zookeeper/service/ZookeeperMetaService.java
+++
b/eventmesh-meta/eventmesh-meta-zookeeper/src/main/java/org/apache/eventmesh/meta/zookeeper/service/ZookeeperMetaService.java
@@ -88,10 +88,10 @@ public class ZookeeperMetaService implements MetaService {
if (null == commonConfiguration) {
continue;
}
- if (StringUtils.isBlank(commonConfiguration.getNamesrvAddr())) {
- throw new MetaException("namesrvAddr cannot be null");
+ if (StringUtils.isBlank(commonConfiguration.getMetaStorageAddr()))
{
+ throw new MetaException("meta storage address cannot be null");
}
- this.serverAddr = commonConfiguration.getNamesrvAddr();
+ this.serverAddr = commonConfiguration.getMetaStorageAddr();
break;
}
ZKRegistryConfiguration zkConfig =
ConfigService.getInstance().buildConfigInstance(ZKRegistryConfiguration.class);
@@ -288,6 +288,21 @@ public class ZookeeperMetaService implements MetaService {
}
}
+ @Override
+ public String getMetaData(String key) {
+ return null;
+ }
+
+ @Override
+ public void updateMetaData(Map<String, String> metadataMap) {
+
+ }
+
+ @Override
+ public void removeMetaData(String key) {
+
+ }
+
@Override
public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo)
throws MetaException {
try {
diff --git
a/eventmesh-meta/eventmesh-meta-zookeeper/src/test/java/org/apache/eventmesh/registry/zookeeper/service/ZookeeperMetaServiceTest.java
b/eventmesh-meta/eventmesh-meta-zookeeper/src/test/java/org/apache/eventmesh/registry/zookeeper/service/ZookeeperMetaServiceTest.java
index ede35a75e..ae03a5685 100644
---
a/eventmesh-meta/eventmesh-meta-zookeeper/src/test/java/org/apache/eventmesh/registry/zookeeper/service/ZookeeperMetaServiceTest.java
+++
b/eventmesh-meta/eventmesh-meta-zookeeper/src/test/java/org/apache/eventmesh/registry/zookeeper/service/ZookeeperMetaServiceTest.java
@@ -61,7 +61,7 @@ public class ZookeeperMetaServiceTest {
zkRegistryService = new ZookeeperMetaService();
CommonConfiguration configuration = new CommonConfiguration();
- configuration.setNamesrvAddr("127.0.0.1:1500");
+ configuration.setMetaStorageAddr("127.0.0.1:1500");
configuration.setEventMeshName("eventmesh");
ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.HTTP,
configuration);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ConfigurationHandler.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ConfigurationHandler.java
index 11220de40..b188440c3 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ConfigurationHandler.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/ConfigurationHandler.java
@@ -113,7 +113,7 @@ public class ConfigurationHandler extends
AbstractHttpHandler {
try {
GetConfigurationResponse getConfigurationResponse = new
GetConfigurationResponse(
eventMeshTCPConfiguration.getSysID(),
- eventMeshTCPConfiguration.getNamesrvAddr(),
+ eventMeshTCPConfiguration.getMetaStorageAddr(),
eventMeshTCPConfiguration.getEventMeshEnv(),
eventMeshTCPConfiguration.getEventMeshIDC(),
eventMeshTCPConfiguration.getEventMeshCluster(),
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcServer.java
index 66aac273a..5cc63f367 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcServer.java
@@ -314,7 +314,7 @@ public class EventMeshGrpcServer {
}
}
- public MetaStorage getRegistry() {
+ public MetaStorage getMetaStorage() {
return metaStorage;
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
index b77699928..4b583c387 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
@@ -128,7 +128,7 @@ public class EventMeshHTTPServer extends AbstractHTTPServer
{
httpRetryer.init();
super.setMetrics(new HTTPMetricsServer(this, metricsRegistries));
- subscriptionManager = new SubscriptionManager();
+ subscriptionManager = new
SubscriptionManager(eventMeshHttpConfiguration.isEventMeshServerMetaStorageEnable(),
metaStorage);
consumerManager = new ConsumerManager(this);
consumerManager.init();
@@ -349,7 +349,7 @@ public class EventMeshHTTPServer extends AbstractHTTPServer
{
return batchRateLimiter;
}
- public MetaStorage getRegistry() {
+ public MetaStorage getMetaStorage() {
return metaStorage;
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
index d5d563940..25319f268 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
@@ -258,7 +258,7 @@ public class EventMeshTCPServer extends AbstractTCPServer {
return eventMeshTCPConfiguration;
}
- public MetaStorage getRegistry() {
+ public MetaStorage getMetaStorage() {
return metaStorage;
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumer/SubscriptionManager.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumer/SubscriptionManager.java
index 37ac2719e..ab5272e9d 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumer/SubscriptionManager.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumer/SubscriptionManager.java
@@ -17,17 +17,24 @@
package org.apache.eventmesh.runtime.core.consumer;
+import org.apache.eventmesh.api.meta.config.EventMeshMetaConfig;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
+import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupMetadata;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicConf;
+import
org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicMetadata;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.Client;
+import org.apache.eventmesh.runtime.meta.MetaStorage;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.Date;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -36,10 +43,19 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class SubscriptionManager {
+ private final boolean isEventMeshServerMetaStorageEnable;
+
+ private final MetaStorage metaStorage;
+
private final ConcurrentHashMap<String /**group*/, ConsumerGroupConf>
localConsumerGroupMapping = new ConcurrentHashMap<>(64);
private final ConcurrentHashMap<String /**group@topic*/, List<Client>>
localClientInfoMapping = new ConcurrentHashMap<>(64);
+ public SubscriptionManager(boolean isEventMeshServerMetaStorageEnable,
MetaStorage metaStorage) {
+ this.isEventMeshServerMetaStorageEnable =
isEventMeshServerMetaStorageEnable;
+ this.metaStorage = metaStorage;
+ }
+
public ConcurrentHashMap<String, ConsumerGroupConf>
getLocalConsumerGroupMapping() {
return localConsumerGroupMapping;
}
@@ -128,4 +144,43 @@ public class SubscriptionManager {
consumerGroupTopicConf.getIdcUrls().get(clientInfo.getIdc()).add(url);
}
}
+
+ public void updateMetaData() {
+ if (!isEventMeshServerMetaStorageEnable) {
+ return;
+ }
+ try {
+ Map<String, String> metadata = new HashMap<>(1 << 4);
+ for (Map.Entry<String, ConsumerGroupConf> consumerGroupMap :
getLocalConsumerGroupMapping().entrySet()) {
+ String consumerGroupKey = consumerGroupMap.getKey();
+ ConsumerGroupConf consumerGroupConf =
consumerGroupMap.getValue();
+
+ ConsumerGroupMetadata consumerGroupMetadata = new
ConsumerGroupMetadata();
+ consumerGroupMetadata.setConsumerGroup(consumerGroupKey);
+
+ Map<String, ConsumerGroupTopicMetadata>
consumerGroupTopicMetadataMap =
+ new HashMap<>(1 << 4);
+ for (Map.Entry<String, ConsumerGroupTopicConf>
consumerGroupTopicConfEntry :
+ consumerGroupConf.getConsumerGroupTopicConf().entrySet()) {
+ final String topic = consumerGroupTopicConfEntry.getKey();
+ ConsumerGroupTopicConf consumerGroupTopicConf =
consumerGroupTopicConfEntry.getValue();
+ ConsumerGroupTopicMetadata consumerGroupTopicMetadata =
new ConsumerGroupTopicMetadata();
+
consumerGroupTopicMetadata.setConsumerGroup(consumerGroupTopicConf.getConsumerGroup());
+
consumerGroupTopicMetadata.setTopic(consumerGroupTopicConf.getTopic());
+
consumerGroupTopicMetadata.setUrls(consumerGroupTopicConf.getUrls());
+
+ consumerGroupTopicMetadataMap.put(topic,
consumerGroupTopicMetadata);
+ }
+
+
consumerGroupMetadata.setConsumerGroupTopicMetadataMap(consumerGroupTopicMetadataMap);
+ metadata.put(consumerGroupKey,
JsonUtils.toJSONString(consumerGroupMetadata));
+ }
+ metadata.put(EventMeshMetaConfig.EVENT_MESH_PROTO, "http");
+
+ metaStorage.updateMetaData(metadata);
+
+ } catch (Exception e) {
+ log.error("update eventmesh metadata error", e);
+ }
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
index fa5929a8b..9496a8913 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
@@ -29,17 +29,24 @@ import
org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.meta.config.EventMeshMetaConfig;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.common.ServiceState;
import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupMetadata;
+import
org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicMetadata;
import org.apache.eventmesh.runtime.core.plugin.MQConsumerWrapper;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupClient;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupTopicConfig;
+import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.GrpcType;
+import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.StreamTopicConfig;
+import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.WebhookTopicConfig;
import
org.apache.eventmesh.runtime.core.protocol.grpc.producer.EventMeshProducer;
import
org.apache.eventmesh.runtime.core.protocol.grpc.producer.SendMessageContext;
import org.apache.eventmesh.runtime.core.protocol.grpc.push.HandleMsgContext;
@@ -48,10 +55,13 @@ import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.commons.collections4.MapUtils;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
@@ -108,7 +118,7 @@ public class EventMeshConsumer {
requireRestart = true;
}
topicConfig.registerClient(client);
-
+ updateMetaData();
return requireRestart;
}
@@ -129,10 +139,61 @@ public class EventMeshConsumer {
requireRestart = true;
}
}
-
+ updateMetaData();
return requireRestart;
}
+ public void updateMetaData() {
+ if (!eventMeshGrpcConfiguration.isEventMeshServerMetaStorageEnable()) {
+ return;
+ }
+ try {
+ Map<String, String> metadata = new HashMap<>(1 << 4);
+ for (Map.Entry<String, ConsumerGroupTopicConfig> consumerGroupMap
: consumerGroupTopicConfig.entrySet()) {
+ String topic = consumerGroupMap.getKey();
+
+ ConsumerGroupTopicConfig cgtConfig =
consumerGroupMap.getValue();
+
+ GrpcType grpcType = cgtConfig.getGrpcType();
+ String consumerGroupKey = cgtConfig.getConsumerGroup();
+ ConsumerGroupMetadata consumerGroupMetadata = new
ConsumerGroupMetadata();
+ Map<String, ConsumerGroupTopicMetadata>
consumerGroupTopicMetadataMap =
+ new HashMap<>(1 << 4);
+ consumerGroupMetadata.setConsumerGroup(consumerGroupKey);
+ if (GrpcType.STREAM == grpcType) {
+ StreamTopicConfig streamTopicConfig = (StreamTopicConfig)
cgtConfig;
+ ConsumerGroupTopicMetadata consumerGroupTopicMetadata =
new ConsumerGroupTopicMetadata();
+
consumerGroupTopicMetadata.setConsumerGroup(streamTopicConfig.getConsumerGroup());
+
consumerGroupTopicMetadata.setTopic(streamTopicConfig.getTopic());
+ Set<String> clientSet = new HashSet<>();
+
streamTopicConfig.getIdcEmitterMap().values().forEach(stringEmitterMap -> {
+ clientSet.addAll(stringEmitterMap.keySet());
+ });
+
+ consumerGroupTopicMetadata.setUrls(clientSet);
+ consumerGroupTopicMetadataMap.put(topic,
consumerGroupTopicMetadata);
+ } else {
+ WebhookTopicConfig webhookTopicConfig =
(WebhookTopicConfig) cgtConfig;
+ ConsumerGroupTopicMetadata consumerGroupTopicMetadata =
new ConsumerGroupTopicMetadata();
+
consumerGroupTopicMetadata.setConsumerGroup(webhookTopicConfig.getConsumerGroup());
+
consumerGroupTopicMetadata.setTopic(webhookTopicConfig.getTopic());
+ Set<String> set = new
HashSet<>(webhookTopicConfig.getTotalUrls());
+ consumerGroupTopicMetadata.setUrls(set);
+ consumerGroupTopicMetadataMap.put(topic,
consumerGroupTopicMetadata);
+ }
+
+
consumerGroupMetadata.setConsumerGroupTopicMetadataMap(consumerGroupTopicMetadataMap);
+ metadata.put(consumerGroupKey,
JsonUtils.toJSONString(consumerGroupMetadata));
+ }
+ metadata.put(EventMeshMetaConfig.EVENT_MESH_PROTO, "grpc");
+
+ eventMeshGrpcServer.getMetaStorage().updateMetaData(metadata);
+
+ } catch (Exception e) {
+ log.error("update eventmesh metadata error", e);
+ }
+ }
+
public synchronized void init() throws Exception {
if (MapUtils.isEmpty(consumerGroupTopicConfig)) {
// no topics, don't init the consumer
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
index f8da24a4e..36632466f 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
@@ -104,6 +104,10 @@ public class StreamTopicConfig extends
ConsumerGroupTopicConfig {
+ ",topic=" + topic + "}";
}
+ public Map<String, Map<String, EventEmitter<CloudEvent>>>
getIdcEmitterMap() {
+ return idcEmitterMap;
+ }
+
public Map<String, List<EventEmitter<CloudEvent>>> getIdcEmitters() {
return idcEmitters;
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
index a0a789bab..10b03d220 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalSubscribeEventProcessor.java
@@ -196,7 +196,7 @@ public class LocalSubscribeEventProcessor extends
AbstractEventProcessor {
}
// Update service metadata
- updateMetadata();
+ eventMeshHTTPServer.getSubscriptionManager().updateMetaData();
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
index 40cb1872e..fc2fe01eb 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/LocalUnSubscribeEventProcessor.java
@@ -232,7 +232,7 @@ public class LocalUnSubscribeEventProcessor extends
AbstractEventProcessor {
}
// Update service metadata
- updateMetadata();
+ eventMeshHTTPServer.getSubscriptionManager().updateMetaData();
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
index d93ba6134..7133f1beb 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
@@ -205,6 +205,7 @@ public class SubscribeProcessor implements
HttpRequestProcessor {
summaryMetrics.recordSendMsgFailed();
summaryMetrics.recordSendMsgCost(endTime - startTime);
}
+ eventMeshHTTPServer.getSubscriptionManager().updateMetaData();
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
index ecc5a0980..85f67a502 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
@@ -234,6 +234,7 @@ public class UnSubscribeProcessor implements
HttpRequestProcessor {
summaryMetrics.recordSendMsgCost(endTime - startTime);
}
}
+ eventMeshHTTPServer.getSubscriptionManager().updateMetaData();
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
index 9ca2403fe..729d3c5f4 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/inf/AbstractEventProcessor.java
@@ -34,9 +34,7 @@ import org.apache.eventmesh.meta.nacos.constant.NacosConstant;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
-import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupMetadata;
-import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicConf;
import
org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicMetadata;
import
org.apache.eventmesh.runtime.core.protocol.http.processor.AsyncHttpProcessor;
import org.apache.eventmesh.runtime.meta.MetaStorage;
@@ -72,52 +70,6 @@ public abstract class AbstractEventProcessor implements
AsyncHttpProcessor {
this.eventMeshHTTPServer = eventMeshHTTPServer;
}
- /**
- * Add a topic with subscribers to the service's metadata.
- */
- protected void updateMetadata() {
- if
(!eventMeshHTTPServer.getEventMeshHttpConfiguration().isEventMeshServerMetaStorageEnable())
{
- return;
- }
-
- try {
-
- Map<String, String> metadata = new HashMap<>(1 << 4);
- for (Map.Entry<String, ConsumerGroupConf> consumerGroupMap :
-
eventMeshHTTPServer.getSubscriptionManager().getLocalConsumerGroupMapping().entrySet())
{
-
- String consumerGroupKey = consumerGroupMap.getKey();
- ConsumerGroupConf consumerGroupConf =
consumerGroupMap.getValue();
-
- ConsumerGroupMetadata consumerGroupMetadata = new
ConsumerGroupMetadata();
- consumerGroupMetadata.setConsumerGroup(consumerGroupKey);
-
- Map<String, ConsumerGroupTopicMetadata>
consumerGroupTopicMetadataMap =
- new HashMap<>(1 << 4);
- for (Map.Entry<String, ConsumerGroupTopicConf>
consumerGroupTopicConfEntry :
- consumerGroupConf.getConsumerGroupTopicConf().entrySet()) {
- final String topic = consumerGroupTopicConfEntry.getKey();
- ConsumerGroupTopicConf consumerGroupTopicConf =
consumerGroupTopicConfEntry.getValue();
- ConsumerGroupTopicMetadata consumerGroupTopicMetadata =
new ConsumerGroupTopicMetadata();
-
consumerGroupTopicMetadata.setConsumerGroup(consumerGroupTopicConf.getConsumerGroup());
-
consumerGroupTopicMetadata.setTopic(consumerGroupTopicConf.getTopic());
-
consumerGroupTopicMetadata.setUrls(consumerGroupTopicConf.getUrls());
-
- consumerGroupTopicMetadataMap.put(topic,
consumerGroupTopicMetadata);
- }
-
-
consumerGroupMetadata.setConsumerGroupTopicMetadataMap(consumerGroupTopicMetadataMap);
- metadata.put(consumerGroupKey,
JsonUtils.toJSONString(consumerGroupMetadata));
- }
-
- eventMeshHTTPServer.getRegistry().registerMetadata(metadata);
-
- } catch (Exception e) {
- log.error("[LocalSubscribeEventProcessor] update eventmesh
metadata error", e);
- }
- }
-
-
protected String getTargetMesh(String consumerGroup,
List<SubscriptionItem> subscriptionList)
throws Exception {
// Currently only supports http
@@ -127,7 +79,7 @@ public abstract class AbstractEventProcessor implements
AsyncHttpProcessor {
}
String targetMesh = "";
- MetaStorage metaStorage = eventMeshHTTPServer.getRegistry();
+ MetaStorage metaStorage = eventMeshHTTPServer.getMetaStorage();
List<EventMeshDataInfo> allEventMeshInfo =
metaStorage.findAllEventMeshInfo();
String httpServiceName =
ConfigurationContextUtil.HTTP + "-" + NacosConstant.GROUP + "@@" +
httpConfiguration.getEventMeshName()
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 7bc038534..3d003bac9 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
@@ -215,7 +215,8 @@ public class ClientGroupWrapper {
Map<String, Session> sessions = new HashMap<>();
topic2sessionInGroupMapping.put(topic, sessions);
}
- if (r =
topic2sessionInGroupMapping.get(topic).putIfAbsent(session.getSessionId(),
session) == null) {
+ Session s =
topic2sessionInGroupMapping.get(topic).putIfAbsent(session.getSessionId(),
session);
+ if (s == null) {
if (log.isInfoEnabled()) {
log.info("Cache session success, group:{} topic:{}
client:{} sessionId:{}", group,
topic, session.getClient(), session.getSessionId());
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
index c86c84e11..394118579 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
@@ -17,12 +17,16 @@
package org.apache.eventmesh.runtime.core.protocol.tcp.client.group;
+import org.apache.eventmesh.api.meta.config.EventMeshMetaConfig;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupMetadata;
+import
org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicMetadata;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcp2Client;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.group.dispatch.DownstreamDispatchStrategy;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.group.dispatch.FreePriorityDispatchStrategy;
@@ -38,8 +42,10 @@ import org.apache.commons.collections4.MapUtils;
import java.lang.ref.WeakReference;
import java.net.InetSocketAddress;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
@@ -482,4 +488,48 @@ public class ClientSessionGroupMapping {
return result;
}
+
+ public void updateMetaData() {
+ if
(!eventMeshTCPServer.getEventMeshTCPConfiguration().isEventMeshServerMetaStorageEnable())
{
+ return;
+ }
+ try {
+ Map<String, String> metadata = new HashMap<>(1 << 4);
+
+ for (Map.Entry<String, ClientGroupWrapper> clientGroupWrapperMap :
this.clientGroupMap.entrySet()) {
+ String group = clientGroupWrapperMap.getKey();
+ ClientGroupWrapper cgw = clientGroupWrapperMap.getValue();
+ ConsumerGroupMetadata consumerGroupMetadata = new
ConsumerGroupMetadata();
+ consumerGroupMetadata.setConsumerGroup(group);
+ Map<String, ConsumerGroupTopicMetadata>
consumerGroupTopicMetadataMap =
+ new HashMap<>(1 << 4);
+ Map<String, Map<String, Session>> topic2sessionInGroupMapping
= cgw.getTopic2sessionInGroupMapping();
+ for (Map.Entry<String, Map<String, Session>> topicSessionMap :
topic2sessionInGroupMapping.entrySet()) {
+ String topic = topicSessionMap.getKey();
+ Map<String, Session> sessionMap =
topicSessionMap.getValue();
+
+ ConsumerGroupTopicMetadata consumerGroupTopicMetadata =
new ConsumerGroupTopicMetadata();
+ consumerGroupTopicMetadata.setConsumerGroup(group);
+ consumerGroupTopicMetadata.setTopic(topic);
+ Set<String> sessionSet = new HashSet<>();
+ for (Map.Entry<String, Session> sessionEntry :
sessionMap.entrySet()) {
+ String sessionId = sessionEntry.getKey();
+ Session session = sessionEntry.getValue();
+ sessionSet.add(session.toString());
+ }
+ consumerGroupTopicMetadata.setUrls(sessionSet);
+ consumerGroupTopicMetadataMap.put(topic,
consumerGroupTopicMetadata);
+ }
+
+
consumerGroupMetadata.setConsumerGroupTopicMetadataMap(consumerGroupTopicMetadataMap);
+ metadata.put(group,
JsonUtils.toJSONString(consumerGroupMetadata));
+ }
+ metadata.put(EventMeshMetaConfig.EVENT_MESH_PROTO, "tcp");
+
+ eventMeshTCPServer.getMetaStorage().updateMetaData(metadata);
+
+ } catch (Exception e) {
+ log.error("update eventmesh metadata error", e);
+ }
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java
index 507ff2ffc..453b6e671 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java
@@ -73,7 +73,7 @@ public class HelloProcessor implements TcpProcessor {
String subsystem = user.getSubsystem();
if
(eventMeshTCPServer.getEventMeshTCPConfiguration().isEventMeshServerSecurityEnable())
{
- EventMeshAppSubTopicInfo eventMeshAppSubTopicInfo =
eventMeshTCPServer.getRegistry().findEventMeshAppSubTopicInfo(group);
+ EventMeshAppSubTopicInfo eventMeshAppSubTopicInfo =
eventMeshTCPServer.getMetaStorage().findEventMeshAppSubTopicInfo(group);
if (eventMeshAppSubTopicInfo == null) {
throw new AclException("no group register");
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java
index 62bf6cc82..7c148f469 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java
@@ -72,7 +72,7 @@ public class SubscribeProcessor implements TcpProcessor {
subscriptionInfo.getTopicList().forEach(item -> {
if (eventMeshServerSecurityEnable) {
try {
- EventMeshAppSubTopicInfo eventMeshAppSubTopicInfo =
eventMeshTCPServer.getRegistry().findEventMeshAppSubTopicInfo(group);
+ EventMeshAppSubTopicInfo eventMeshAppSubTopicInfo =
eventMeshTCPServer.getMetaStorage().findEventMeshAppSubTopicInfo(group);
if (eventMeshAppSubTopicInfo == null) {
throw new AclException("no group register");
}
@@ -91,6 +91,7 @@ public class SubscribeProcessor implements TcpProcessor {
log.info("SubscribeTask succeed|user={}|topics={}",
session.getClient(), subscriptionItems);
}
}
+ eventMeshTCPServer.getClientSessionGroupMapping().updateMetaData();
msg.setHeader(new Header(Command.SUBSCRIBE_RESPONSE,
OPStatus.SUCCESS.getCode(), OPStatus.SUCCESS.getDesc(),
pkg.getHeader().getSeq()));
} catch (Exception e) {
log.error("SubscribeTask failed|user={}|errMsg={}",
session.getClient(), e);
@@ -100,5 +101,4 @@ public class SubscribeProcessor implements TcpProcessor {
}
}
-
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceImpl.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceImpl.java
index 15d6cf857..eb2f13596 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceImpl.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceImpl.java
@@ -79,7 +79,7 @@ public class EventMeshRebalanceImpl implements
EventMeshRebalanceStrategy {
Map<String, String> localEventMeshMap = null;
List<EventMeshDataInfo> eventMeshDataInfoList = null;
try {
- eventMeshDataInfoList =
eventMeshTCPServer.getRegistry().findEventMeshInfoByCluster(cluster);
+ eventMeshDataInfoList =
eventMeshTCPServer.getMetaStorage().findEventMeshInfoByCluster(cluster);
if (eventMeshDataInfoList == null ||
CollectionUtils.isEmpty(eventMeshDataInfoList)) {
log.warn("doRebalance failed,query eventmesh instances is null
from registry,cluster:{}", cluster);
@@ -222,7 +222,7 @@ public class EventMeshRebalanceImpl implements
EventMeshRebalanceStrategy {
Map<String, Integer> localEventMeshDistributeData = null;
Map<String, Map<String, Integer>> eventMeshClientDistributionDataMap =
null;
try {
- eventMeshClientDistributionDataMap =
eventMeshTCPServer.getRegistry().findEventMeshClientDistributionData(
+ eventMeshClientDistributionDataMap =
eventMeshTCPServer.getMetaStorage().findEventMeshClientDistributionData(
cluster, group, purpose);
if (MapUtils.isEmpty(eventMeshClientDistributionDataMap)) {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
index 087e5b63c..5ca1415de 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
@@ -58,7 +58,7 @@ public class EventMeshRecommendImpl implements
EventMeshRecommendStrategy {
final String cluster =
eventMeshTCPServer.getEventMeshTCPConfiguration().getEventMeshCluster();
try {
- eventMeshDataInfoList =
eventMeshTCPServer.getRegistry().findEventMeshInfoByCluster(cluster);
+ eventMeshDataInfoList =
eventMeshTCPServer.getMetaStorage().findEventMeshInfoByCluster(cluster);
} catch (Exception e) {
if (log.isWarnEnabled()) {
log.warn("EventMeshRecommend failed,
findEventMeshInfoByCluster failed, cluster:{}, group:{}, purpose:{}, errMsg:{}",
@@ -162,7 +162,7 @@ public class EventMeshRecommendImpl implements
EventMeshRecommendStrategy {
Map<String, Map<String, Integer>> eventMeshClientDistributionDataMap =
null;
try {
- eventMeshClientDistributionDataMap =
eventMeshTCPServer.getRegistry().findEventMeshClientDistributionData(
+ eventMeshClientDistributionDataMap =
eventMeshTCPServer.getMetaStorage().findEventMeshClientDistributionData(
cluster, group, purpose);
} catch (Exception e) {
if (log.isWarnEnabled()) {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/meta/MetaStorage.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/meta/MetaStorage.java
index 9756fe19a..71904e4b8 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/meta/MetaStorage.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/meta/MetaStorage.java
@@ -110,6 +110,10 @@ public class MetaStorage {
metaService.registerMetadata(metadata);
}
+ public void updateMetaData(Map<String, String> metadata) {
+ metaService.updateMetaData(metadata);
+ }
+
public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo)
throws MetaException {
return metaService.register(eventMeshRegisterInfo);
}
diff --git a/eventmesh-runtime/src/test/resources/configuration.properties
b/eventmesh-runtime/src/test/resources/configuration.properties
index 59c1e2944..71ad34e19 100644
--- a/eventmesh-runtime/src/test/resources/configuration.properties
+++ b/eventmesh-runtime/src/test/resources/configuration.properties
@@ -25,20 +25,21 @@ eventMesh.server.hostIp=hostIp-succeed!!!
eventMesh.connector.plugin.type=connector-succeed!!!
eventMesh.storage.plugin.type=storage-succeed!!!
eventMesh.security.plugin.type=security-succeed!!!
-eventMesh.metaStorage.plugin.type=metaStorage-succeed!!!
eventMesh.trace.plugin=trace-succeed!!!
-eventMesh.server.metaStorage.registerIntervalInMills=816
-eventMesh.server.metaStorage.fetchRegistryAddrIntervalInMills=1816
+
eventMesh.metrics.plugin=metrics-succeed1!!!,metrics-succeed2!!!,metrics-succeed3!!!
+eventMesh.metaStorage.plugin.type=metaStorage-succeed!!!
eventMesh.metaStorage.plugin.server-addr=server-addr-succeed1!!!
+eventMesh.metaStorage.plugin.enabled=true
+eventMesh.metaStorage.plugin.username=username-succeed!!!
+eventMesh.metaStorage.plugin.password=password-succeed!!!
+eventMesh.metaStorage.plugin.registerIntervalInMills=816
+eventMesh.metaStorage.plugin.fetchRegistryAddrIntervalInMills=1816
eventMesh.server.security.enabled=true
-eventMesh.server.metaStorage.enabled=true
eventMesh.server.trace.enabled=true
eventMesh.server.provide.protocols=TCP,HTTP,GRPC
-eventMesh.metaStorage.plugin.username=username-succeed!!!
-eventMesh.metaStorage.plugin.password=password-succeed!!!
# EventMeshHTTPConfiguration config
eventMesh.server.http.port=1816
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
