apache · dockerzhang · Mar 14, 2022 · Mar 14, 2022
diff --git a/...rver/src/main/java/org/apache/inlong/tubemq/server/master/metamanage/MetaDataManager.java b/...rver/src/main/java/org/apache/inlong/tubemq/server/master/metamanage/MetaDataManager.java
@@ -47,7 +47,7 @@
 import org.apache.inlong.tubemq.server.master.MasterConfig;
 import org.apache.inlong.tubemq.server.master.TMaster;
 import org.apache.inlong.tubemq.server.master.bdbstore.MasterGroupStatus;
-import org.apache.inlong.tubemq.server.master.metamanage.keepalive.AliveObserver;
+import org.apache.inlong.tubemq.server.master.metamanage.metastore.MetaConfigObserver;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.impl.bdbimpl.BdbMetaStoreServiceImpl;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.MetaStoreService;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.BaseEntity;
@@ -144,7 +144,7 @@ public void stop() throws Exception {
         logger.info("BrokerConfManager StoreService stopped");
     }
 
-    public void registerObserver(AliveObserver eventObserver) {
+    public void registerObserver(MetaConfigObserver eventObserver) {
         metaStoreService.registerObserver(eventObserver);
     }
 

diff --git a/...aster/metamanage/keepalive/KeepAlive.java → ...etamanage/metastore/KeepAliveService.java b/...aster/metamanage/keepalive/KeepAlive.java → ...etamanage/metastore/KeepAliveService.java
@@ -15,42 +15,63 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.tubemq.server.master.metamanage.keepalive;
+package org.apache.inlong.tubemq.server.master.metamanage.metastore;
 
 import java.net.InetSocketAddress;
+
+import org.apache.inlong.tubemq.server.Server;
 import org.apache.inlong.tubemq.server.master.bdbstore.MasterGroupStatus;
 import org.apache.inlong.tubemq.server.master.web.model.ClusterGroupVO;
 
-public interface KeepAlive {
+public interface KeepAliveService extends Server {
 
     /**
-     * Whether this node is the master role
+     * Whether this node is the Master role now
      *
-     * @return true if is master role or else
+     * @return true if is Master role or else
      */
     boolean isMasterNow();
 
+    /**
+     * Get the timestamp when the current node becomes the Master role
+     *
+     * @return the since time
+     */
     long getMasterSinceTime();
 
+    /**
+     * Get current Master address
+     *
+     * @return  the current Master address
+     */
     InetSocketAddress getMasterAddress();
 
     /**
-     * Whether the primary node in active
+     * Whether the Master node in active
      *
-     * @return  true for active, false for inactive
+     * @return  true for Master, false for Slave
      */
     boolean isPrimaryNodeActive();
 
+    /**
+     * Transfer Master role to other replica node
+     *
+     * @throws Exception the exception information
+     */
     void transferMaster() throws Exception;
 
     /**
-     * Register node role switching event observer
+     * Get group address info
      *
-     * @param eventObserver  the event observer
+     * @return  the group address information
      */
-    void registerObserver(AliveObserver eventObserver);
-
     ClusterGroupVO getGroupAddressStrInfo();
 
+    /**
+     * Get Master group status
+     *
+     * @param isFromHeartbeat   whether called by hb thread
+     * @return    the master group status
+     */
     MasterGroupStatus getMasterGroupStatus(boolean isFromHeartbeat);
 }
diff --git a/...r/metamanage/keepalive/AliveObserver.java → ...amanage/metastore/MetaConfigObserver.java b/...r/metamanage/keepalive/AliveObserver.java → ...amanage/metastore/MetaConfigObserver.java
@@ -15,9 +15,9 @@
  * limitations under the License.
  */
 
-package org.apache.inlong.tubemq.server.master.metamanage.keepalive;
+package org.apache.inlong.tubemq.server.master.metamanage.metastore;
 
-public interface AliveObserver {
+public interface MetaConfigObserver {
 
     void clearCacheData();
 

diff --git a/...in/java/org/apache/inlong/tubemq/server/master/metamanage/metastore/MetaStoreService.java b/...in/java/org/apache/inlong/tubemq/server/master/metamanage/metastore/MetaStoreService.java
@@ -22,16 +22,14 @@
 import java.util.Set;
 import org.apache.inlong.tubemq.corebase.rv.ProcessResult;
 import org.apache.inlong.tubemq.server.Server;
-import org.apache.inlong.tubemq.server.master.metamanage.keepalive.AliveObserver;
-import org.apache.inlong.tubemq.server.master.metamanage.keepalive.KeepAlive;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.BrokerConfEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.ClusterSettingEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.GroupConsumeCtrlEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.GroupResCtrlEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.TopicCtrlEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.TopicDeployEntity;
 
-public interface MetaStoreService extends KeepAlive, Server {
+public interface MetaStoreService extends KeepAliveService, Server {
 
     boolean checkStoreStatus(boolean checkIsMaster, ProcessResult result);
 
@@ -279,7 +277,7 @@ boolean delGroupConsumeCtrlConf(String operator,
     boolean delGroupConsumeCtrlConf(String operator, String recordKey,
                                     StringBuilder strBuff, ProcessResult result);
 
-    void registerObserver(AliveObserver eventObserver);
+    void registerObserver(MetaConfigObserver eventObserver);
 
     boolean isTopicNameInUsed(String topicName);
 

diff --git a/...metastore/dao/mapper/MetaStoreMapper.java → ...etastore/dao/mapper/MetaConfigMapper.java b/...metastore/dao/mapper/MetaStoreMapper.java → ...etastore/dao/mapper/MetaConfigMapper.java
@@ -23,6 +23,8 @@
 import org.apache.inlong.tubemq.corebase.rv.ProcessResult;
 import org.apache.inlong.tubemq.server.common.statusdef.ManageStatus;
 import org.apache.inlong.tubemq.server.common.statusdef.TopicStatus;
+import org.apache.inlong.tubemq.server.master.metamanage.metastore.MetaConfigObserver;
+import org.apache.inlong.tubemq.server.master.metamanage.metastore.KeepAliveService;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.BaseEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.BrokerConfEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.ClusterSettingEntity;
@@ -32,7 +34,16 @@
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.TopicDeployEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.TopicPropGroup;
 
-public interface MetaStoreMapper {
+public interface MetaConfigMapper extends KeepAliveService {
+
+    /**
+     * Register meta configure change observer
+     *
+     * @param eventObserver  the event observer
+     */
+    void regMetaConfigObserver(MetaConfigObserver eventObserver);
+
+    boolean checkStoreStatus(boolean checkIsMaster, ProcessResult result);
 
     /**
      * Add or update cluster default setting

diff --git a/...etastore/impl/AbsMetaStoreMapperImpl.java → ...tastore/impl/AbsMetaConfigMapperImpl.java b/...etastore/impl/AbsMetaStoreMapperImpl.java → ...tastore/impl/AbsMetaConfigMapperImpl.java
@@ -23,7 +23,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.commons.codec.binary.StringUtils;
 import org.apache.inlong.tubemq.corebase.TBaseConstants;
 import org.apache.inlong.tubemq.corebase.rv.ProcessResult;
@@ -32,8 +31,8 @@
 import org.apache.inlong.tubemq.server.common.statusdef.TopicStatus;
 import org.apache.inlong.tubemq.server.common.utils.RowLock;
 import org.apache.inlong.tubemq.server.master.metamanage.DataOpErrCode;
-import org.apache.inlong.tubemq.server.master.metamanage.keepalive.AliveObserver;
-import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.mapper.MetaStoreMapper;
+import org.apache.inlong.tubemq.server.master.metamanage.metastore.MetaConfigObserver;
+import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.mapper.MetaConfigMapper;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.BaseEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.BrokerConfEntity;
 import org.apache.inlong.tubemq.server.master.metamanage.metastore.dao.entity.ClusterSettingEntity;
@@ -51,38 +50,41 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class AbsMetaStoreMapperImpl implements MetaStoreMapper {
+public abstract class AbsMetaConfigMapperImpl implements MetaConfigMapper {
     protected static final Logger logger =
-            LoggerFactory.getLogger(AbsMetaStoreMapperImpl.class);
-    // service status
-    // 0 stopped, 1 starting, 2 started, 3 stopping
-    private final AtomicInteger srvStatus = new AtomicInteger(0);
-    // the observers focusing on active-standby switching
-    private final List<AliveObserver> eventObservers = new ArrayList<>();
-
+            LoggerFactory.getLogger(AbsMetaConfigMapperImpl.class);
     // row lock.
     private final RowLock metaRowLock;
     // default cluster setting
     private static final ClusterSettingEntity defClusterSetting =
             new ClusterSettingEntity().fillDefaultValue();
     // cluster default setting
-    private ClusterConfigMapper clusterConfigMapper;
+    protected ClusterConfigMapper clusterConfigMapper;
     // broker configure
-    private BrokerConfigMapper brokerConfigMapper;
+    protected BrokerConfigMapper brokerConfigMapper;
     // topic deployment configure
-    private TopicDeployMapper topicDeployMapper;
+    protected TopicDeployMapper topicDeployMapper;
     // topic control configure
-    private TopicCtrlMapper topicCtrlMapper;
+    protected TopicCtrlMapper topicCtrlMapper;
     // group resource control configure
-    private GroupResCtrlMapper groupResCtrlMapper;
+    protected GroupResCtrlMapper groupResCtrlMapper;
     // group consume control configure
-    private ConsumeCtrlMapper consumeCtrlMapper;
+    protected ConsumeCtrlMapper consumeCtrlMapper;
+    // the observers focusing on active-standby switching
+    private final List<MetaConfigObserver> eventObservers = new ArrayList<>();
 
-    public AbsMetaStoreMapperImpl(int rowLockWaiDurMs) {
+    public AbsMetaConfigMapperImpl(int rowLockWaiDurMs) {
         this.metaRowLock =
                 new RowLock("MetaData-RowLock", rowLockWaiDurMs);
     }
 
+    @Override
+    public void regMetaConfigObserver(MetaConfigObserver eventObserver) {
+        if (eventObserver != null) {
+            eventObservers.add(eventObserver);
+        }
+    }
+
     @Override
     public boolean addOrUpdClusterDefSetting(BaseEntity opEntity,
                                              int brokerPort, int brokerTlsPort,
@@ -1156,22 +1158,14 @@ public Map<String, List<GroupConsumeCtrlEntity>> getGroupConsumeCtrlConf(
         return consumeCtrlMapper.getConsumeCtrlInfoMap(groupSet, topicSet, qryEntry);
     }
 
-    /**
-     * Initial meta-data stores.
-     *
-     */
-    protected void initMetaStore() {
-
-    }
-
     /**
      * Reload meta-data stores.
      *
      * @param strBuff  the string buffer
      */
-    private void reloadMetaStore(StringBuilder strBuff) {
+    protected void reloadMetaStore(StringBuilder strBuff) {
         // Clear observers' cache data.
-        for (AliveObserver observer : eventObservers) {
+        for (MetaConfigObserver observer : eventObservers) {
             observer.clearCacheData();
         }
         // Load the latest meta-data from persistent
@@ -1182,7 +1176,7 @@ private void reloadMetaStore(StringBuilder strBuff) {
         groupResCtrlMapper.loadConfig(strBuff);
         consumeCtrlMapper.loadConfig(strBuff);
         // load the latest meta-data to observers
-        for (AliveObserver observer : eventObservers) {
+        for (MetaConfigObserver observer : eventObservers) {
             observer.reloadCacheData();
         }
     }
@@ -1191,7 +1185,7 @@ private void reloadMetaStore(StringBuilder strBuff) {
      * Close meta-data stores.
      *
      */
-    private void closeMetaStore() {
+    protected void closeMetaStore() {
         brokerConfigMapper.close();
         topicDeployMapper.close();
         groupResCtrlMapper.close();