This is an automated email from the ASF dual-hosted git repository.
gosonzhang pushed a commit to branch TUBEMQ-570
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/TUBEMQ-570 by this push:
new 797da0a [INLONG-611] Add FSM for broker configure manage
797da0a is described below
commit 797da0a69b77d35609d9e2714e14be56799a5c7e
Author: gosonzhang <[email protected]>
AuthorDate: Sun May 9 11:00:49 2021 +0800
[INLONG-611] Add FSM for broker configure manage
---
.../client/consumer/ConsumerSamplePrint.java | 9 +
.../tubemq/corebase/utils/AbstractSamplePrint.java | 5 +
.../org/apache/tubemq/corebase/utils/Tuple2.java | 3 +-
.../org/apache/tubemq/corebase/utils/Tuple3.java | 10 +-
.../corebase/utils/{Tuple3.java => Tuple4.java} | 26 +-
.../server/broker/utils/BrokerSamplePrint.java | 9 +
.../server/broker/utils/DiskSamplePrint.java | 9 +
.../tubemq/server/common/statusdef/StepStatus.java | 64 ++++
.../org/apache/tubemq/server/master/TMaster.java | 18 +-
.../server/master/metamanage/MetaDataManager.java | 57 +++-
.../nodemanage/nodebroker/BrokerPSInfoHolder.java | 181 ++++++++++
.../nodemanage/nodebroker/BrokerRunManager.java | 59 ++++
.../nodemanage/nodebroker/BrokerRunStatusInfo.java | 357 ++++++++++++++++++++
.../nodemanage/nodebroker/BrokerSyncData.java | 322 ++++++++++++++++++
.../nodemanage/nodebroker/BrokerTopicInfoView.java | 372 +++++++++++++++++++++
.../nodemanage/nodebroker/TargetValidResult.java | 32 --
.../server/master/utils/BdbStoreSamplePrint.java | 9 +
.../master/utils/BrokerStatusSamplePrint.java | 82 +++++
18 files changed, 1556 insertions(+), 68 deletions(-)
diff --git
a/tubemq-client/src/main/java/org/apache/tubemq/client/consumer/ConsumerSamplePrint.java
b/tubemq-client/src/main/java/org/apache/tubemq/client/consumer/ConsumerSamplePrint.java
index 0aadf2e..8d6911f 100644
---
a/tubemq-client/src/main/java/org/apache/tubemq/client/consumer/ConsumerSamplePrint.java
+++
b/tubemq-client/src/main/java/org/apache/tubemq/client/consumer/ConsumerSamplePrint.java
@@ -71,4 +71,13 @@ public class ConsumerSamplePrint extends AbstractSamplePrint
{
//
}
+ @Override
+ public void printWarn(String err) {
+ //
+ }
+
+ @Override
+ public void printError(String err) {
+ //
+ }
}
diff --git
a/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/AbstractSamplePrint.java
b/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/AbstractSamplePrint.java
index f009245..34f132f 100644
---
a/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/AbstractSamplePrint.java
+++
b/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/AbstractSamplePrint.java
@@ -51,4 +51,9 @@ public abstract class AbstractSamplePrint {
public abstract void printExceptionCaught(Throwable e);
public abstract void printExceptionCaught(Throwable e, String hostName,
String nodeName);
+
+ public abstract void printWarn(String err);
+
+ public abstract void printError(String err);
+
}
diff --git
a/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple2.java
b/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple2.java
index c9f16e5..5f2af74 100644
--- a/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple2.java
+++ b/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple2.java
@@ -47,8 +47,7 @@ public class Tuple2<T0, T1> {
* @param value1 The value for field 1
*/
public Tuple2(T0 value0, T1 value1) {
- this.f0 = value0;
- this.f1 = value1;
+ setF0AndF1(value0, value1);
}
public T0 getF0() {
diff --git
a/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple3.java
b/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple3.java
index 579b425..b7f744a 100644
--- a/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple3.java
+++ b/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple3.java
@@ -41,9 +41,7 @@ public class Tuple3<T0, T1, T2> {
* @param value2 The value for field 2
*/
public Tuple3(T0 value0, T1 value1, T2 value2) {
- this.f0 = value0;
- this.f1 = value1;
- this.f2 = value2;
+ setFieldsValue(value0, value1, value2);
}
public T0 getF0() {
@@ -57,4 +55,10 @@ public class Tuple3<T0, T1, T2> {
public T2 getF2() {
return f2;
}
+
+ public void setFieldsValue(T0 value0, T1 value1, T2 value2) {
+ this.f0 = value0;
+ this.f1 = value1;
+ this.f2 = value2;
+ }
}
diff --git
a/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple3.java
b/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple4.java
similarity index 75%
copy from tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple3.java
copy to tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple4.java
index 579b425..67ace7a 100644
--- a/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple3.java
+++ b/tubemq-core/src/main/java/org/apache/tubemq/corebase/utils/Tuple4.java
@@ -17,7 +17,7 @@
package org.apache.tubemq.corebase.utils;
-public class Tuple3<T0, T1, T2> {
+public class Tuple4<T0, T1, T2, T3> {
/** Field 0 of the tuple. */
private T0 f0 = null;
@@ -25,11 +25,12 @@ public class Tuple3<T0, T1, T2> {
private T1 f1 = null;
/** Field 2 of the tuple. */
private T2 f2 = null;
-
+ /** Field 3 of the tuple. */
+ private T3 f3 = null;
/**
* Creates a new tuple where all fields are null.
*/
- public Tuple3() {
+ public Tuple4() {
}
@@ -39,11 +40,10 @@ public class Tuple3<T0, T1, T2> {
* @param value0 The value for field 0
* @param value1 The value for field 1
* @param value2 The value for field 2
+ * @param value3 The value for field 3
*/
- public Tuple3(T0 value0, T1 value1, T2 value2) {
- this.f0 = value0;
- this.f1 = value1;
- this.f2 = value2;
+ public Tuple4(T0 value0, T1 value1, T2 value2, T3 value3) {
+ setFieldsValue(value0, value1, value2, value3);
}
public T0 getF0() {
@@ -57,4 +57,16 @@ public class Tuple3<T0, T1, T2> {
public T2 getF2() {
return f2;
}
+
+ public T3 getF3() {
+ return f3;
+ }
+
+ public void setFieldsValue(T0 value0, T1 value1, T2 value2, T3 value3) {
+ this.f0 = value0;
+ this.f1 = value1;
+ this.f2 = value2;
+ this.f3 = value3;
+ }
+
}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/broker/utils/BrokerSamplePrint.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/broker/utils/BrokerSamplePrint.java
index d0c32a1..54f2ceb 100644
---
a/tubemq-server/src/main/java/org/apache/tubemq/server/broker/utils/BrokerSamplePrint.java
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/broker/utils/BrokerSamplePrint.java
@@ -80,4 +80,13 @@ public class BrokerSamplePrint extends AbstractSamplePrint {
//
}
+ @Override
+ public void printWarn(String err) {
+ //
+ }
+
+ @Override
+ public void printError(String err) {
+ //
+ }
}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/broker/utils/DiskSamplePrint.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/broker/utils/DiskSamplePrint.java
index a1c29c0..2f4ac23 100644
---
a/tubemq-server/src/main/java/org/apache/tubemq/server/broker/utils/DiskSamplePrint.java
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/broker/utils/DiskSamplePrint.java
@@ -117,4 +117,13 @@ public class DiskSamplePrint extends AbstractSamplePrint {
}
}
+ @Override
+ public void printWarn(String err) {
+ //
+ }
+
+ @Override
+ public void printError(String err) {
+ //
+ }
}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/common/statusdef/StepStatus.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/common/statusdef/StepStatus.java
new file mode 100644
index 0000000..cfa4038
--- /dev/null
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/common/statusdef/StepStatus.java
@@ -0,0 +1,64 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.tubemq.server.common.statusdef;
+
+
+public enum StepStatus {
+
+ STEP_STATUS_UNDEFINED(-2, "idle", 0),
+ STEP_STATUS_LOAD_DATA(1, "load_data", 0),
+ STEP_STATUS_WAIT_ONLINE(2, "wait_online", 0),
+ STEP_STATUS_WAIT_SYNC(3, "wait_sync", 0),
+ STEP_STATUS_WAIT_SUBSCRIBE(4, "wait_sub", 60000),
+ STEP_STATUS_WAIT_PUBLISH(5, "wait_pub", 30000);
+
+ private int code;
+ private String description;
+ private long delayDurInMs;
+
+
+
+ StepStatus(int code, String description, long delayDurInMs) {
+ this.code = code;
+ this.description = description;
+ this.delayDurInMs = delayDurInMs;
+ }
+
+ public int getCode() {
+ return code;
+ }
+
+ public String getDescription() {
+ return description;
+ }
+
+ public long getDelayDurInMs() {
+ return delayDurInMs;
+ }
+
+ public static StepStatus valueOf(int code) {
+ for (StepStatus status : StepStatus.values()) {
+ if (status.getCode() == code) {
+ return status;
+ }
+ }
+ throw new IllegalArgumentException(String.format(
+ "unknown broker step status code %s", code));
+ }
+
+}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/TMaster.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/TMaster.java
index 651e655..c8f99e4 100644
--- a/tubemq-server/src/main/java/org/apache/tubemq/server/master/TMaster.java
+++ b/tubemq-server/src/main/java/org/apache/tubemq/server/master/TMaster.java
@@ -1014,15 +1014,20 @@ public class TMaster extends HasThread implements
MasterService, Stoppable {
ConcurrentHashMap<Integer, BrokerSyncStatusInfo> brokerSyncStatusMap =
this.defMetaDataManager.getBrokerRunSyncManageMap();
// update broker status
- List<String> brokerTopicSetConfInfo =
+ Map<String, String> brokerTopicSetConfInfo =
this.defMetaDataManager.getBrokerTopicStrConfigInfo(brokerConfEntity,
strBuffer);
+ List<String> brokerTopicSetConfInfoList =
+ new ArrayList<>(brokerTopicSetConfInfo.size());
+ for (String topicItem : brokerTopicSetConfInfo.values()) {
+ brokerTopicSetConfInfoList.add(topicItem);
+ }
BrokerSyncStatusInfo brokerStatusInfo =
- new BrokerSyncStatusInfo(brokerConfEntity,
brokerTopicSetConfInfo);
+ new BrokerSyncStatusInfo(brokerConfEntity,
brokerTopicSetConfInfoList);
brokerSyncStatusMap.put(brokerConfEntity.getBrokerId(),
brokerStatusInfo);
brokerStatusInfo.updateCurrBrokerConfInfo(brokerConfEntity.getManageStatus().getCode(),
brokerConfEntity.isConfDataUpdated(),
brokerConfEntity.isBrokerLoaded(),
- brokerConfEntity.getBrokerDefaultConfInfo(),
brokerTopicSetConfInfo, false);
- if (brokerTopicSetConfInfo.isEmpty()) {
+ brokerConfEntity.getBrokerDefaultConfInfo(),
brokerTopicSetConfInfoList, false);
+ if (brokerTopicSetConfInfoList.isEmpty()) {
needFastStart = true;
}
brokerStatusInfo.setFastStart(needFastStart);
@@ -1035,7 +1040,7 @@ public class TMaster extends HasThread implements
MasterService, Stoppable {
brokerStatusInfo.setBrokerReportInfo(true,
brokerStatusInfo.getLastPushBrokerConfId(),
brokerStatusInfo.getLastPushBrokerCheckSumId(), true,
- brokerConfEntity.getBrokerDefaultConfInfo(),
brokerTopicSetConfInfo, true,
+ brokerConfEntity.getBrokerDefaultConfInfo(),
brokerTopicSetConfInfoList, true,
request.getBrokerOnline(), overtls);
}
this.defMetaDataManager.removeBrokerRunTopicInfoMap(brokerInfo.getBrokerId());
@@ -1197,6 +1202,7 @@ public class TMaster extends HasThread implements
MasterService, Stoppable {
builder.setErrMsg(e.getMessage());
return builder.build();
}
+
// update broker status
brokerSyncStatusInfo.setBrokerReportInfo(false,
request.getCurBrokerConfId(),
request.getConfCheckSumId(), request.getTakeConfInfo(),
@@ -1260,7 +1266,7 @@ public class TMaster extends HasThread implements
MasterService, Stoppable {
}
builder.setTakeRemoveTopicInfo(true);
builder.addAllRemoveTopicConfInfo(defMetaDataManager
- .getBrokerRemovedTopicStrConfigInfo(brokerConfigEntry,
strBuffer));
+ .getBrokerRemovedTopicStrConfigInfo(brokerConfigEntry,
strBuffer).values());
if (brokerSyncStatusInfo.needSyncConfDataToBroker()) {
builder.setTakeConfInfo(true);
builder.setBrokerDefaultConfInfo(brokerSyncStatusInfo
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/metamanage/MetaDataManager.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/metamanage/MetaDataManager.java
index 1250f83..a34a13a 100644
---
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/metamanage/MetaDataManager.java
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/metamanage/MetaDataManager.java
@@ -147,10 +147,15 @@ public class MetaDataManager implements Server {
boolean needFastStart = false;
BrokerSyncStatusInfo brokerSyncStatusInfo =
this.brokerRunSyncManageMap.get(entity.getBrokerId());
- List<String> brokerTopicSetConfInfo =
getBrokerTopicStrConfigInfo(entity, sBuffer);
+ Map<String, String> brokerTopicSetConfInfo =
getBrokerTopicStrConfigInfo(entity, sBuffer);
+ List<String> brokerTopicSetConfInfoList =
+ new ArrayList<>(brokerTopicSetConfInfo.size());
+ for (String topicItem : brokerTopicSetConfInfo.values()) {
+ brokerTopicSetConfInfoList.add(topicItem);
+ }
if (brokerSyncStatusInfo == null) {
brokerSyncStatusInfo =
- new BrokerSyncStatusInfo(entity,
brokerTopicSetConfInfo);
+ new BrokerSyncStatusInfo(entity,
brokerTopicSetConfInfoList);
BrokerSyncStatusInfo tmpBrokerSyncStatusInfo =
brokerRunSyncManageMap.putIfAbsent(entity.getBrokerId(),
brokerSyncStatusInfo);
@@ -164,7 +169,7 @@ public class MetaDataManager implements Server {
brokerSyncStatusInfo.setFastStart(needFastStart);
brokerSyncStatusInfo.updateCurrBrokerConfInfo(entity.getManageStatus().getCode(),
entity.isConfDataUpdated(), entity.isBrokerLoaded(),
- entity.getBrokerDefaultConfInfo(),
brokerTopicSetConfInfo, false);
+ entity.getBrokerDefaultConfInfo(),
brokerTopicSetConfInfoList, false);
}
}
isStarted = true;
@@ -866,13 +871,18 @@ public class MetaDataManager implements Server {
return result.isSuccess();
}
String curBrokerConfStr = curEntity.getBrokerDefaultConfInfo();
- List<String> curBrokerTopicConfStrSet =
+ Map<String, String> curBrokerTopicConfStrSet =
getBrokerTopicStrConfigInfo(curEntity, strBuffer);
+ List<String> brokerTopicSetConfInfoList =
+ new ArrayList<>(curBrokerTopicConfStrSet.size());
+ for (String topicItem : curBrokerTopicConfStrSet.values()) {
+ brokerTopicSetConfInfoList.add(topicItem);
+ }
BrokerSyncStatusInfo brokerSyncStatusInfo =
this.brokerRunSyncManageMap.get(entity.getBrokerId());
if (brokerSyncStatusInfo == null) {
brokerSyncStatusInfo =
- new BrokerSyncStatusInfo(entity, curBrokerTopicConfStrSet);
+ new BrokerSyncStatusInfo(entity,
brokerTopicSetConfInfoList);
BrokerSyncStatusInfo tmpBrokerSyncStatusInfo =
brokerRunSyncManageMap.putIfAbsent(entity.getBrokerId(),
brokerSyncStatusInfo);
if (tmpBrokerSyncStatusInfo != null) {
@@ -901,7 +911,7 @@ public class MetaDataManager implements Server {
|| oldManageStatus ==
TStatusConstants.STATUS_MANAGE_ONLINE_NOT_READ);
brokerSyncStatusInfo.updateCurrBrokerConfInfo(curManageStatus,
curEntity.isConfDataUpdated(), curEntity.isBrokerLoaded(),
curBrokerConfStr,
- curBrokerTopicConfStrSet, isOnlineUpdate);
+ brokerTopicSetConfInfoList, isOnlineUpdate);
} else {
brokerSyncStatusInfo.setBrokerOffline();
}
@@ -1085,10 +1095,15 @@ public class MetaDataManager implements Server {
BrokerSyncStatusInfo brokerSyncStatusInfo =
brokerRunSyncManageMap.get(curEntity.getBrokerId());
if (brokerSyncStatusInfo == null) {
- List<String> newBrokerTopicConfStrSet =
+ Map<String, String> newBrokerTopicConfStrSet =
getBrokerTopicStrConfigInfo(curEntity, strBuffer);
+ List<String> brokerTopicSetConfInfoList =
+ new ArrayList<>(newBrokerTopicConfStrSet.size());
+ for (String topicItem : newBrokerTopicConfStrSet.values())
{
+ brokerTopicSetConfInfoList.add(topicItem);
+ }
brokerSyncStatusInfo =
- new BrokerSyncStatusInfo(curEntity,
newBrokerTopicConfStrSet);
+ new BrokerSyncStatusInfo(curEntity,
brokerTopicSetConfInfoList);
BrokerSyncStatusInfo tmpBrokerSyncStatusInfo =
brokerRunSyncManageMap.putIfAbsent(curEntity.getBrokerId(),
brokerSyncStatusInfo);
if (tmpBrokerSyncStatusInfo != null) {
@@ -1111,10 +1126,15 @@ public class MetaDataManager implements Server {
BrokerSyncStatusInfo brokerSyncStatusInfo =
brokerRunSyncManageMap.get(curEntity.getBrokerId());
if (brokerSyncStatusInfo == null) {
- List<String> newBrokerTopicConfStrSet =
+ Map<String, String> newBrokerTopicConfStrSet =
getBrokerTopicStrConfigInfo(curEntity, strBuffer);
+ List<String> brokerTopicSetConfInfoList =
+ new ArrayList<>(newBrokerTopicConfStrSet.size());
+ for (String topicItem : newBrokerTopicConfStrSet.values())
{
+ brokerTopicSetConfInfoList.add(topicItem);
+ }
brokerSyncStatusInfo =
- new BrokerSyncStatusInfo(curEntity,
newBrokerTopicConfStrSet);
+ new BrokerSyncStatusInfo(curEntity,
brokerTopicSetConfInfoList);
BrokerSyncStatusInfo tmpBrokerSyncStatusInfo =
brokerRunSyncManageMap.putIfAbsent(curEntity.getBrokerId(),
brokerSyncStatusInfo);
@@ -1446,23 +1466,24 @@ public class MetaDataManager implements Server {
recordKey, strBuffer, result);
}
- public List<String> getBrokerTopicStrConfigInfo(
+ public Map<String, String> getBrokerTopicStrConfigInfo(
BrokerConfEntity brokerConfEntity, StringBuilder sBuffer) {
return inGetTopicConfStrInfo(brokerConfEntity, false, sBuffer);
}
- public List<String> getBrokerRemovedTopicStrConfigInfo(
+ public Map<String, String> getBrokerRemovedTopicStrConfigInfo(
BrokerConfEntity brokerConfEntity, StringBuilder sBuffer) {
return inGetTopicConfStrInfo(brokerConfEntity, true, sBuffer);
}
- private List<String> inGetTopicConfStrInfo(BrokerConfEntity brokerEntity,
- boolean isRemoved,
StringBuilder sBuffer) {
- List<String> topicConfStrs = new ArrayList<>();
+ private Map<String, String> inGetTopicConfStrInfo(BrokerConfEntity
brokerEntity,
+ boolean isRemoved,
+ StringBuilder sBuffer) {
+ Map<String, String> topicConfStrMap = new HashMap<>();
Map<String, TopicDeployEntity> topicEntityMap =
metaStoreService.getConfiguredTopicInfo(brokerEntity.getBrokerId());
if (topicEntityMap.isEmpty()) {
- return topicConfStrs;
+ return topicConfStrMap;
}
TopicPropGroup defTopicProps = brokerEntity.getTopicProps();
ClusterSettingEntity clusterDefConf =
@@ -1549,10 +1570,10 @@ public class MetaDataManager implements Server {
} else {
sBuffer.append(TokenConstants.ATTR_SEP).append(maxMsgSize);
}
- topicConfStrs.add(sBuffer.toString());
+ topicConfStrMap.put(topicEntity.getTopicName(),
sBuffer.toString());
sBuffer.delete(0, sBuffer.length());
}
- return topicConfStrs;
+ return topicConfStrMap;
}
//
/////////////////////////////////////////////////////////////////////////////////
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerPSInfoHolder.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerPSInfoHolder.java
new file mode 100644
index 0000000..3fd9cd4
--- /dev/null
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerPSInfoHolder.java
@@ -0,0 +1,181 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.tubemq.server.master.nodemanage.nodebroker;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.tubemq.corebase.cluster.Partition;
+import org.apache.tubemq.corebase.cluster.TopicInfo;
+import org.apache.tubemq.corebase.utils.ConcurrentHashSet;
+import org.apache.tubemq.corebase.utils.Tuple2;
+import org.apache.tubemq.server.common.statusdef.ManageStatus;
+
+
+public class BrokerPSInfoHolder {
+ // broker manage status
+ private ConcurrentHashSet<Integer/* brokerId */> enablePubBrokerIdSet =
new ConcurrentHashSet<>();
+ private ConcurrentHashSet<Integer/* brokerId */> enableSubBrokerIdSet =
new ConcurrentHashSet<>();
+ // broker subscribe topic view info
+ private BrokerTopicInfoView subTopicInfoView = new BrokerTopicInfoView();
+ // broker publish topic view info
+ private BrokerTopicInfoView pubTopicInfoView = new BrokerTopicInfoView();
+
+
+ public BrokerPSInfoHolder() {
+
+ }
+
+ /**
+ * remove broker all configure info
+ *
+ * @param brokerId broker id index
+ */
+ public void rmvBrokerAllPushedInfo(int brokerId) {
+ // remove broker status Info
+ enablePubBrokerIdSet.remove(brokerId);
+ enableSubBrokerIdSet.remove(brokerId);
+ // remove broker topic info
+ subTopicInfoView.rmvBrokerTopicInfo(brokerId);
+ pubTopicInfoView.rmvBrokerTopicInfo(brokerId);
+ }
+
+ /**
+ * update broker manage status
+ *
+ * @param brokerId broker id index
+ * @param mngStatus broker's manage status
+ */
+ public void updBrokerMangeStatus(int brokerId, ManageStatus mngStatus) {
+ Tuple2<Boolean, Boolean> pubSubStatus = mngStatus.getPubSubStatus();
+ if (pubSubStatus.getF0() == Boolean.TRUE) {
+ enablePubBrokerIdSet.add(brokerId);
+ } else {
+ enablePubBrokerIdSet.remove(brokerId);
+ }
+ if (pubSubStatus.getF1() == Boolean.TRUE) {
+ enableSubBrokerIdSet.add(brokerId);
+ } else {
+ enableSubBrokerIdSet.remove(brokerId);
+ }
+ }
+
+ /**
+ * update broker's subscribe topicInfo configures
+ *
+ * @param brokerId broker id index
+ * @param topicInfoMap broker's topic configure info,
+ * if topicInfoMap is null, reserve current configure;
+ * if topicInfoMap is empty, clear current configure.
+ */
+ public void updBrokerSubTopicConfInfo(int brokerId,
+ Map<String, TopicInfo> topicInfoMap)
{
+ if (topicInfoMap == null) {
+ return;
+ }
+ subTopicInfoView.updBrokerTopicConfInfo(brokerId, topicInfoMap);
+ }
+
+ /**
+ * update broker's publish topicInfo configures
+ *
+ * @param brokerId broker id index
+ * @param topicInfoMap broker's topic configure info,
+ * if topicInfoMap is null, reserve current configure;
+ * if topicInfoMap is empty, clear current configure.
+ */
+ public void updBrokerPubTopicConfInfo(int brokerId,
+ Map<String, TopicInfo> topicInfoMap)
{
+ if (topicInfoMap == null) {
+ return;
+ }
+ pubTopicInfoView.updBrokerTopicConfInfo(brokerId, topicInfoMap);
+ }
+
+ /**
+ * update broker manage status and topicInfo configures
+ *
+ * @param brokerId broker id index
+ * @param mngStatus broker's manage status
+ * @param topicInfoMap broker's topic configure info
+ */
+ public void updateBrokerPushedInfo(int brokerId, ManageStatus mngStatus,
+ Map<String, TopicInfo> topicInfoMap) {
+ updBrokerMangeStatus(brokerId, mngStatus);
+ updBrokerSubTopicConfInfo(brokerId, topicInfoMap);
+ updBrokerPubTopicConfInfo(brokerId, topicInfoMap);
+ }
+
+ /**
+ * Get the maximum number of broker distributions of topic
+ *
+ * @param topicSet need query topic set
+ */
+ public int getTopicMaxSubBrokerCnt(Set<String> topicSet) {
+ return subTopicInfoView.getMaxTopicBrokerCnt(topicSet);
+ }
+
+ /**
+ * Gets the list of topic partitions whose subscribe status is enabled
+ *
+ * @param topicSet need query topic set
+ */
+ public List<Partition> getAcceptSubParts(Set<String> topicSet) {
+ return subTopicInfoView.getAcceptSubParts(topicSet,
enableSubBrokerIdSet);
+ }
+
+ /**
+ * Gets the list of topic partitions whose subscribe status is enabled
+ *
+ * @param topic need query topic set
+ */
+ public List<Partition> getAcceptSubParts(String topic) {
+ return subTopicInfoView.getAcceptSubParts(topic, enableSubBrokerIdSet);
+ }
+
+ /**
+ * Gets the string map of topic partitions whose publish status is enabled
+ *
+ * @param topicSet need query topic set
+ */
+ public Map<String, String> getAcceptPubPartInfo(Set<String> topicSet) {
+ return pubTopicInfoView.getAcceptPubPartInfo(topicSet,
enablePubBrokerIdSet);
+ }
+
+ /**
+ * Get the published TopicInfo information of topic in broker
+ *
+ * @param brokerId need query broker
+ * @param topic need query topic
+ *
+ * @return null or topicInfo configure
+ */
+ public TopicInfo getBrokerPubPushedTopicInfo(int brokerId, String topic) {
+ return pubTopicInfoView.getBrokerPushedTopicInfo(brokerId, topic);
+ }
+
+ /**
+ * Get all published TopicInfo information of broker
+ *
+ * @param brokerId need query broker
+ */
+ public List<TopicInfo> getPubBrokerPushedTopicInfo(int brokerId) {
+ return pubTopicInfoView.getBrokerPushedTopicInfo(brokerId);
+ }
+
+}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerRunManager.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerRunManager.java
new file mode 100644
index 0000000..b37e37a
--- /dev/null
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerRunManager.java
@@ -0,0 +1,59 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.tubemq.server.master.nodemanage.nodebroker;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.tubemq.corebase.cluster.TopicInfo;
+import org.apache.tubemq.corebase.utils.Tuple3;
+import org.apache.tubemq.server.common.statusdef.ManageStatus;
+import org.apache.tubemq.server.common.utils.ProcessResult;
+
+
+public interface BrokerRunManager {
+
+ boolean brokerRegister2M(String clientId, boolean enableTls,
+ int tlsPort, long reportConfigId,
+ int reportCheckSumId, boolean isTackData,
+ String repBrokerConfInfo,
+ List<String> repTopicConfInfo,
+ boolean isOnline, boolean isOverTLS,
+ StringBuilder sBuffer, ProcessResult result);
+
+ boolean brokerHeartBeat2M(int brokerId, long reportConfigId,
+ int reportCheckSumId, boolean isTackData,
+ String repBrokerConfInfo,
+ List<String> repTopicConfInfo, boolean isOnline,
+ StringBuilder sBuffer, ProcessResult result);
+
+ boolean brokerClose2M(int brokerId);
+
+ boolean brokerTimeout(int brokerId, long bookedId);
+
+
+ Tuple3<ManageStatus, String, Map<String, String>>
getBrokerMetaConfigInfo(int brokerId);
+
+ void updBrokerCsmConfInfo(int brokerId,
+ ManageStatus mngStatus,
+ Map<String, TopicInfo> topicInfoMap);
+
+ void updBrokerPrdConfInfo(int brokerId,
+ ManageStatus mngStatus,
+ Map<String, TopicInfo> topicInfoMap);
+
+}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerRunStatusInfo.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerRunStatusInfo.java
new file mode 100644
index 0000000..45c5459
--- /dev/null
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerRunStatusInfo.java
@@ -0,0 +1,357 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.tubemq.server.master.nodemanage.nodebroker;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.tubemq.corebase.TBaseConstants;
+import org.apache.tubemq.corebase.cluster.BrokerInfo;
+import org.apache.tubemq.corebase.cluster.TopicInfo;
+import org.apache.tubemq.corebase.utils.TStringUtils;
+import org.apache.tubemq.corebase.utils.Tuple2;
+import org.apache.tubemq.corebase.utils.Tuple3;
+import org.apache.tubemq.corebase.utils.Tuple4;
+import org.apache.tubemq.server.common.TServerConstants;
+import org.apache.tubemq.server.common.statusdef.ManageStatus;
+import org.apache.tubemq.server.common.statusdef.StepStatus;
+import org.apache.tubemq.server.master.utils.BrokerStatusSamplePrint;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class BrokerRunStatusInfo {
+
+ private static final Logger logger =
+ LoggerFactory.getLogger(BrokerRunStatusInfo.class);
+ private static final BrokerStatusSamplePrint statusFsmSamplePrint =
+ new BrokerStatusSamplePrint(logger);
+
+ private final BrokerRunManager brokerRunManager;
+
+ private BrokerInfo brokerInfo;
+ private long createId;
+ // config change flag
+ private AtomicBoolean isConfChanged = new AtomicBoolean(false);
+ private AtomicLong confChangeNo =
+ new AtomicLong(TBaseConstants.META_VALUE_UNDEFINED);
+ // data sync status
+ private StepStatus curStepStatus;
+ private volatile long nextStepOpTimeInMills = 0;
+ // current loaded change No.
+ private AtomicLong confLoadedNo =
+ new AtomicLong(TBaseConstants.META_VALUE_UNDEFINED);
+ // broker sync data info
+ BrokerSyncData brokerSyncData = new BrokerSyncData();
+ // broker status conditions
+ private boolean isOnline = false; // broker online flag
+ private boolean isDoneDataLoad = false;
+ private boolean isDoneDataSub = false;
+ private boolean isDoneDataPub = false;
+ private boolean isOverTLS = false; // enable tls
+ private long lastBrokerSyncTime = 0;
+ private long maxConfLoadedTimeInMs = 0;
+ private long curConfLoadTimeInMs = 0;
+
+
+
+ public BrokerRunStatusInfo(BrokerRunManager brokerRunManager, BrokerInfo
brokerInfo,
+ ManageStatus mngStatus, String brokerConfInfo,
+ Map<String, String> topicConfInfoMap, boolean
isOverTls) {
+ this.brokerRunManager = brokerRunManager;
+ reInitRunStatusInfo(brokerInfo, mngStatus,
+ brokerConfInfo, topicConfInfoMap, isOverTls);
+ }
+
+ public void reInitRunStatusInfo(BrokerInfo brokerInfo, ManageStatus
mngStatus,
+ String brokerConfInfo, Map<String, String>
topicConfInfoMap,
+ boolean isOverTls) {
+ resetStatusInfo();
+ this.createId = System.nanoTime();
+ this.brokerInfo = brokerInfo;
+ long curTime = System.currentTimeMillis();
+ this.isOverTLS = isOverTls;
+ this.isConfChanged.set(true);
+ this.confChangeNo.set(curTime);
+ this.brokerSyncData.updBrokerSyncData(true,
+ confChangeNo.get(), mngStatus, brokerConfInfo,
topicConfInfoMap);
+ this.curStepStatus = StepStatus.STEP_STATUS_WAIT_ONLINE;
+ this.nextStepOpTimeInMills = curTime + curStepStatus.getDelayDurInMs();
+ }
+
+ public void notifyDataChanged() {
+ this.isConfChanged.set(true);
+ this.confChangeNo.incrementAndGet();
+ }
+
+ public Tuple2<Boolean, Boolean> getDataSyncStatus() {
+ return new Tuple2<>(this.isConfChanged.get(),
+ this.confChangeNo.get() == this.confLoadedNo.get());
+ }
+
+ private boolean isDataChanged() {
+ return (this.isConfChanged.get()
+ || (this.curStepStatus == StepStatus.STEP_STATUS_UNDEFINED
+ && this.confChangeNo.get() != this.confLoadedNo.get()));
+ }
+
+ public long getCreateId() {
+ return this.createId;
+ }
+
+
+ /**
+ * Get need sync to broker's data
+ *
+ * @param retValue return data container, ** must not null **
+ * @return void
+ */
+ public void getNeedSyncData(Tuple4<Long,
+ Integer, String, Map<String, String>> retValue) {
+ brokerSyncData.getBrokerSyncData(retValue);
+ }
+
+ /**
+ * Book broker report info
+ *
+ * @param isRegister request method
+ * @param isOnline if broker is ready
+ * @param repConfigId report configure id
+ * @param repCheckSumId report configure check sum
+ * @param isTackData if tack data in request
+ * @param repBrokerConfInfo tacked broker configure
+ * @param repTopicConfs tacked topic configure
+ * @param sBuffer string process container
+ */
+ public void bookBrokerReportInfo(boolean isRegister, boolean isOnline,
+ long repConfigId, int repCheckSumId,
+ boolean isTackData, String
repBrokerConfInfo,
+ List<String> repTopicConfs,
+ StringBuilder sBuffer) {
+ boolean isSynchronized =
+ brokerSyncData.bookBrokerReportInfo(repConfigId,
+ repCheckSumId, isTackData, repBrokerConfInfo,
repTopicConfs);
+ this.isOnline = isOnline;
+ goNextStatus(isRegister, isSynchronized, sBuffer);
+ }
+
+ private void goNextStatus(boolean isRegister,
+ boolean isSynchronized,
+ StringBuilder sBuffer) {
+ if (isRegister) {
+ goRegNextStatus(isSynchronized);
+ } else {
+ goNonRegNextStatus(isSynchronized);
+ }
+ execEvent(sBuffer);
+ }
+
+ private void execEvent(StringBuilder sBuffer) {
+ switch (curStepStatus) {
+ case STEP_STATUS_LOAD_DATA: {
+ loadNewMetaData(sBuffer);
+ }
+ break;
+
+ case STEP_STATUS_WAIT_SUBSCRIBE: {
+ execSyncDataToSub();
+ }
+ break;
+
+ case STEP_STATUS_WAIT_PUBLISH: {
+ execSyncDataToPub();
+ }
+ break;
+
+ case STEP_STATUS_UNDEFINED:
+ case STEP_STATUS_WAIT_ONLINE:
+ case STEP_STATUS_WAIT_SYNC:
+ default: {
+ }
+ }
+ }
+
+ private void goRegNextStatus(boolean isSynchronized) {
+ resetStatusInfo();
+ if (isOnline) {
+ if (isSynchronized) {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_SUBSCRIBE;
+ } else {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_SYNC;
+ }
+ } else {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_ONLINE;
+ }
+ }
+
+ private void goNonRegNextStatus(boolean isSynchronized) {
+ switch (curStepStatus) {
+ case STEP_STATUS_UNDEFINED: {
+ if (!isSynchronized || isDataChanged() || needForceSyncData())
{
+ resetStatusInfo();
+ curStepStatus = StepStatus.STEP_STATUS_LOAD_DATA;
+ nextStepOpTimeInMills =
+ System.currentTimeMillis() +
curStepStatus.getDelayDurInMs();
+ }
+ }
+ break;
+
+ case STEP_STATUS_LOAD_DATA: {
+ if (isDoneDataLoad) {
+ if (isOnline) {
+ if (isSynchronized) {
+ curStepStatus =
StepStatus.STEP_STATUS_WAIT_SUBSCRIBE;
+ } else {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_SYNC;
+ }
+ nextStepOpTimeInMills =
+ System.currentTimeMillis() +
curStepStatus.getDelayDurInMs();
+ }
+ }
+ }
+ break;
+
+ case STEP_STATUS_WAIT_ONLINE: {
+ if (isOnline) {
+ if (isSynchronized) {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_SUBSCRIBE;
+ } else {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_SYNC;
+ }
+ nextStepOpTimeInMills =
+ System.currentTimeMillis() +
curStepStatus.getDelayDurInMs();
+ }
+ }
+ break;
+
+ case STEP_STATUS_WAIT_SYNC: {
+ if (isSynchronized) {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_SUBSCRIBE;
+ } else {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_SYNC;
+ }
+ nextStepOpTimeInMills =
+ System.currentTimeMillis() +
curStepStatus.getDelayDurInMs();
+ }
+ break;
+
+ case STEP_STATUS_WAIT_SUBSCRIBE: {
+ if (isDoneDataSub && System.currentTimeMillis() >
nextStepOpTimeInMills) {
+ curStepStatus = StepStatus.STEP_STATUS_WAIT_PUBLISH;
+ nextStepOpTimeInMills =
+ System.currentTimeMillis() +
curStepStatus.getDelayDurInMs();
+ }
+ }
+ break;
+
+ case STEP_STATUS_WAIT_PUBLISH:
+ default: {
+ if (isDoneDataPub && System.currentTimeMillis() >
nextStepOpTimeInMills) {
+ finishedDataSync();
+ }
+ }
+ }
+ }
+
+ private void loadNewMetaData(StringBuilder sBuffer) {
+ if (isDoneDataLoad) {
+ return;
+ }
+ boolean needForceSync = needForceSyncData();
+ Tuple3<ManageStatus, String, Map<String, String>> curConfTuple =
+
brokerRunManager.getBrokerMetaConfigInfo(brokerInfo.getBrokerId());
+ if (TStringUtils.isBlank(curConfTuple.getF1())) {
+ statusFsmSamplePrint.printWarn(sBuffer
+ .append("[Broker Sync] found broker(")
+ .append(brokerInfo.getBrokerId())
+ .append(") configure is null").toString());
+ sBuffer.delete(0, sBuffer.length());
+ return;
+ }
+ Tuple2<Boolean, Boolean> updResult =
+ brokerSyncData.updBrokerSyncData(needForceSync,
+ confChangeNo.get(), curConfTuple.getF0(),
+ curConfTuple.getF1(), curConfTuple.getF2());
+ if (updResult == null) {
+ return;
+ }
+ isDoneDataLoad = true;
+ this.lastBrokerSyncTime = System.currentTimeMillis();
+ if (!updResult.getF0() && !updResult.getF1()) {
+ finishedDataSync();
+ }
+ }
+
+ private void execSyncDataToSub() {
+ if (isDoneDataSub) {
+ return;
+ }
+ Tuple2<ManageStatus, Map<String, TopicInfo>> syncData =
+ brokerSyncData.getBrokerPublishInfo(brokerInfo);
+ brokerRunManager.updBrokerCsmConfInfo(brokerInfo.getBrokerId(),
+ syncData.getF0(), syncData.getF1());
+ isDoneDataSub = true;
+ }
+
+ private void execSyncDataToPub() {
+ if (isDoneDataPub) {
+ return;
+ }
+ Tuple2<ManageStatus, Map<String, TopicInfo>> syncData =
+ brokerSyncData.getBrokerPublishInfo(brokerInfo);
+ brokerRunManager.updBrokerPrdConfInfo(brokerInfo.getBrokerId(),
+ syncData.getF0(), syncData.getF1());
+ isDoneDataPub = true;
+ }
+
+ private void finishedDataSync() {
+ this.confLoadedNo.set(brokerSyncData.getDataPushId());
+ if (this.confLoadedNo.get() == this.confChangeNo.get()) {
+ this.isConfChanged.set(false);
+ }
+ curStepStatus = StepStatus.STEP_STATUS_UNDEFINED;
+ nextStepOpTimeInMills = 0;
+ long tmpDuration = System.currentTimeMillis() - curConfLoadTimeInMs;
+ if (maxConfLoadedTimeInMs < tmpDuration) {
+ maxConfLoadedTimeInMs = tmpDuration;
+ }
+ }
+
+ private void resetStatusInfo() {
+ isDoneDataLoad = false;
+ isDoneDataSub = false;
+ isDoneDataPub = false;
+ nextStepOpTimeInMills = 0;
+ curConfLoadTimeInMs = System.currentTimeMillis();
+ }
+
+ /**
+ * According to last report time and current time to decide if need to
report data
+ *
+ * @return true if need report data otherwise false
+ */
+ private boolean needForceSyncData() {
+ if (System.currentTimeMillis() - this.lastBrokerSyncTime
+ > TServerConstants.CFG_REPORT_DEFAULT_SYNC_DURATION) {
+ return true;
+ }
+ return false;
+ }
+
+}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerSyncData.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerSyncData.java
new file mode 100644
index 0000000..09de48d
--- /dev/null
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerSyncData.java
@@ -0,0 +1,322 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.tubemq.server.master.nodemanage.nodebroker;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.commons.codec.binary.StringUtils;
+import org.apache.tubemq.corebase.TBaseConstants;
+import org.apache.tubemq.corebase.TokenConstants;
+import org.apache.tubemq.corebase.cluster.BrokerInfo;
+import org.apache.tubemq.corebase.cluster.TopicInfo;
+import org.apache.tubemq.corebase.utils.CheckSum;
+import org.apache.tubemq.corebase.utils.TStringUtils;
+import org.apache.tubemq.corebase.utils.Tuple2;
+import org.apache.tubemq.corebase.utils.Tuple4;
+import org.apache.tubemq.server.common.TServerConstants;
+import org.apache.tubemq.server.common.statusdef.ManageStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+
+public class BrokerSyncData {
+ private static final Logger logger =
+ LoggerFactory.getLogger(BrokerSyncData.class);
+ // current data push id
+ private long dataPushId;
+ // data need to sync
+ private AtomicLong syncDownDataConfId =
+ new AtomicLong(System.currentTimeMillis());
+ private int syncDownDataChkSumId;
+ private ManageStatus mngStatus;
+ private String syncDownBrokerConfInfo;
+ private Map<String, String> syncDownTopicConfInfoMap = new HashMap<>();
+ private boolean isStatusChanged = false;
+ private boolean isConfChanged = false;
+
+ // report info
+ private long syncUpDataConfId = TBaseConstants.META_VALUE_UNDEFINED;
+ private int syncUpDataChkSumId = TBaseConstants.META_VALUE_UNDEFINED;
+ private String syncUpBrokerConfInfo;
+ private List<String> syncUpTopicConfInfos = new ArrayList<>();
+ private long lastDataUpTime = 0;
+
+ public BrokerSyncData() {
+
+ }
+
+ public BrokerSyncData(long dataPushId,
+ ManageStatus mngStatus,
+ String brokerConfInfo,
+ Map<String, String> topicConfInfoMap) {
+ updBrokerSyncData(true, dataPushId,
+ mngStatus, brokerConfInfo, topicConfInfoMap);
+ }
+
+ /**
+ * Update current sync data
+ *
+ * @param isForceSync if force sync data to broker
+ * @param dataPushId data push id for verify
+ * @param mngStatus new broker manage status
+ * @param brokerConfInfo broker default configure
+ * @param topicConfInfoMap topic configure set
+ *
+ * @return whether changed, if null: input brokerConfInfo data is illegal
+ * f0 : manage status changed
+ * f1 : configure changed
+ */
+ public Tuple2<Boolean, Boolean> updBrokerSyncData(boolean isForceSync,
long dataPushId,
+ ManageStatus mngStatus,
+ String brokerConfInfo,
+ Map<String, String>
topicConfInfoMap) {
+ isStatusChanged = false;
+ isConfChanged = false;
+ // check parameters
+ if (TStringUtils.isBlank(brokerConfInfo)) {
+ return null;
+ }
+ this.dataPushId = dataPushId;
+ if (isForceSync || this.mngStatus != mngStatus) {
+ this.mngStatus = mngStatus;
+ isStatusChanged = true;
+ }
+
+ if (isForceSync || isSyncDataChanged(brokerConfInfo,
topicConfInfoMap)) {
+ this.syncDownBrokerConfInfo = brokerConfInfo;
+ if (topicConfInfoMap == null) {
+ this.syncDownTopicConfInfoMap = new HashMap<>();
+ } else {
+ this.syncDownTopicConfInfoMap = topicConfInfoMap;
+ }
+ this.syncDownDataChkSumId =
+ calculateConfigCrc32Value(syncDownBrokerConfInfo,
syncDownTopicConfInfoMap);
+ isConfChanged = true;
+ }
+ if (isStatusChanged && isConfChanged) {
+ this.syncDownDataConfId.incrementAndGet();
+ }
+ return new Tuple2<>(isStatusChanged, isConfChanged);
+ }
+
+ /**
+ * Book the report data by broker
+ * @param syncDataConfId data configure id
+ * @param syncDataChkSumId data check-sum id
+ * @param isTakeData if carry the data info
+ * @param syncBrokerConfInfo broker default config
+ * @param syncTopicConfInfos topic config set
+ *
+ * @return whether the broker data synchronized
+ */
+ public boolean bookBrokerReportInfo(long syncDataConfId, int
syncDataChkSumId,
+ boolean isTakeData, String
syncBrokerConfInfo,
+ List<String> syncTopicConfInfos) {
+ this.syncUpDataConfId = syncDataConfId;
+ this.syncUpDataChkSumId = syncDataChkSumId;
+ if (isTakeData) {
+ this.syncUpBrokerConfInfo = syncBrokerConfInfo;
+ if (syncTopicConfInfos == null) {
+ this.syncUpTopicConfInfos = new ArrayList<>();
+ } else {
+ this.syncUpTopicConfInfos = syncTopicConfInfos;
+ }
+ this.lastDataUpTime = System.currentTimeMillis();
+ }
+ return isConfSynchronized();
+ }
+
+ /**
+ * whether the broker data is synchronized
+ *
+ * @return true: the configure is synchronized;
+ * false: not synchronized
+ */
+ public boolean isConfSynchronized() {
+ return (this.syncDownDataConfId.get() == this.syncUpDataConfId
+ && this.syncDownDataChkSumId == this.syncUpDataChkSumId);
+ }
+
+ public boolean isDownTopicConfEmpty() {
+ return this.syncDownTopicConfInfoMap.isEmpty();
+ }
+
+ /**
+ * Get need sync to broker's data
+ *
+ * @param retValue return data container, ** must not null **
+ * @return void
+ */
+ public void getBrokerSyncData(Tuple4<Long,
+ Integer, String, Map<String, String>> retValue) {
+ if (isConfSynchronized()) {
+ retValue.setFieldsValue(syncDownDataConfId.get(),
syncDownDataChkSumId,
+ syncDownBrokerConfInfo, syncDownTopicConfInfoMap);
+ } else {
+ retValue.setFieldsValue(syncDownDataConfId.get(),
+ syncDownDataChkSumId, null, null);
+ }
+ }
+
+ /**
+ * Get the broker publish info
+ * @param brokerInfo broker info
+ * @return need sync data
+ * f0 : manage status
+ * f1 : topic configure
+ */
+ public Tuple2<ManageStatus, Map<String, TopicInfo>> getBrokerPublishInfo(
+ final BrokerInfo brokerInfo) {
+ Map<String, TopicInfo> topicInfoMap = parseTopicInfoConf(brokerInfo);
+ return new Tuple2<>(mngStatus, topicInfoMap);
+ }
+
+ public long getDataPushId() {
+ return dataPushId;
+ }
+
+ public long getLastDataReportTime() {
+ return lastDataUpTime;
+ }
+
+ /**
+ * Judge the sync-data is change
+ * @param brokerConfInfo broker default config
+ * @param topicConfInfoMap topic config set
+ *
+ * @return whether the sync-data is change
+ */
+ private boolean isSyncDataChanged(String brokerConfInfo,
+ Map<String, String> topicConfInfoMap) {
+ return !Objects.equals(syncDownBrokerConfInfo, brokerConfInfo)
+ || !Objects.equals(syncDownTopicConfInfoMap, topicConfInfoMap);
+ }
+
+ /**
+ * parse broker report configure to topicInfo
+ *
+ * @param brokerInfo broker info
+ */
+ private Map<String, TopicInfo> parseTopicInfoConf(final BrokerInfo
brokerInfo) {
+ // check broker configure
+ if (TStringUtils.isBlank(this.syncUpBrokerConfInfo)) {
+ return null;
+ }
+ // get broker status and topic default configure
+ String[] brokerConfInfoAttrs =
+ this.syncUpBrokerConfInfo.split(TokenConstants.ATTR_SEP);
+ int numPartitions = Integer.parseInt(brokerConfInfoAttrs[0]);
+ boolean cfgAcceptPublish =
Boolean.parseBoolean(brokerConfInfoAttrs[1]);
+ boolean cfgAcceptSubscribe =
Boolean.parseBoolean(brokerConfInfoAttrs[2]);
+ int numTopicStores = 1;
+ if (brokerConfInfoAttrs.length > 7) {
+ if (!TStringUtils.isBlank(brokerConfInfoAttrs[7])) {
+ numTopicStores = Integer.parseInt(brokerConfInfoAttrs[7]);
+ }
+ }
+ int unFlushDataHold = TServerConstants.CFG_DEFAULT_DATA_UNFLUSH_HOLD;
+ if (brokerConfInfoAttrs.length > 8) {
+ if (!TStringUtils.isBlank(brokerConfInfoAttrs[8])) {
+ unFlushDataHold = Integer.parseInt(brokerConfInfoAttrs[8]);
+ }
+ }
+ Map<String, TopicInfo> topicInfoMap = new HashMap<>();
+ // make up topicInfo info according to broker default configure
+ for (String strTopicConfInfo : this.syncUpTopicConfInfos) {
+ if (TStringUtils.isBlank(strTopicConfInfo)) {
+ continue;
+ }
+ String[] topicConfAttrs =
+ strTopicConfInfo.split(TokenConstants.ATTR_SEP);
+ final String tmpTopic = topicConfAttrs[0];
+ int tmpPartNum = numPartitions;
+ if (!TStringUtils.isBlank(topicConfAttrs[1])) {
+ tmpPartNum = Integer.parseInt(topicConfAttrs[1]);
+ }
+ boolean tmpAcceptPublish = cfgAcceptPublish;
+ if (!TStringUtils.isBlank(topicConfAttrs[2])) {
+ tmpAcceptPublish = Boolean.parseBoolean(topicConfAttrs[2]);
+ }
+ int tmpNumTopicStores = numTopicStores;
+ if (!TStringUtils.isBlank(topicConfAttrs[8])) {
+ tmpNumTopicStores = Integer.parseInt(topicConfAttrs[8]);
+ tmpNumTopicStores = tmpNumTopicStores > 0 ? tmpNumTopicStores
: numTopicStores;
+ }
+ boolean tmpAcceptSubscribe = cfgAcceptSubscribe;
+ if (!TStringUtils.isBlank(topicConfAttrs[3])) {
+ tmpAcceptSubscribe = Boolean.parseBoolean(topicConfAttrs[3]);
+ }
+ topicInfoMap.put(tmpTopic, new TopicInfo(brokerInfo, tmpTopic,
+ tmpPartNum, tmpNumTopicStores, tmpAcceptPublish,
tmpAcceptSubscribe));
+ }
+ return topicInfoMap;
+ }
+
+ /**
+ * Calculate the sync-data's crc32 value
+ *
+ * @param brokerConfInfo broker default config
+ * @param topicConfInfoMap topic config set
+ *
+ * @return the crc32 value
+ */
+ private int calculateConfigCrc32Value(String brokerConfInfo,
+ Map<String, String>
topicConfInfoMap) {
+ int result = -1;
+ int capacity = 0;
+ List<String> topicConfInfoLst =
+ new ArrayList<>(topicConfInfoMap.values());
+ Collections.sort(topicConfInfoLst);
+ capacity += brokerConfInfo.length();
+ for (String itemStr : topicConfInfoLst) {
+ capacity += itemStr.length();
+ }
+ capacity *= 2;
+ for (int i = 1; i < 3; i++) {
+ result = inCalcBufferResult(capacity, brokerConfInfo,
topicConfInfoLst);
+ if (result >= 0) {
+ return result;
+ }
+ capacity *= i + 1;
+ }
+ logger.error("Calculate the CRC32 value of Broker Configure error!");
+ return 0;
+ }
+
+ private int inCalcBufferResult(int capacity, String brokerConfInfo,
+ List<String> topicConfInfoLst) {
+ final ByteBuffer buffer = ByteBuffer.allocate(capacity);
+ buffer.put(StringUtils.getBytesUtf8(brokerConfInfo));
+ for (String itemStr : topicConfInfoLst) {
+ byte[] itemData = StringUtils.getBytesUtf8(itemStr);
+ if (itemData.length > buffer.remaining()) {
+ return -1;
+ }
+ buffer.put(itemData);
+ }
+ return CheckSum.crc32(buffer.array());
+ }
+
+}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerTopicInfoView.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerTopicInfoView.java
new file mode 100644
index 0000000..ddfca1e
--- /dev/null
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/BrokerTopicInfoView.java
@@ -0,0 +1,372 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.tubemq.server.master.nodemanage.nodebroker;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.tubemq.corebase.TBaseConstants;
+import org.apache.tubemq.corebase.TokenConstants;
+import org.apache.tubemq.corebase.cluster.Partition;
+import org.apache.tubemq.corebase.cluster.TopicInfo;
+import org.apache.tubemq.corebase.utils.ConcurrentHashSet;
+
+public class BrokerTopicInfoView {
+ private ConcurrentHashMap<String/* topicName */, TopicInfoView>
+ topicConfInfoMap = new ConcurrentHashMap<>();
+ private ConcurrentHashMap<Integer/* brokerId */,
ConcurrentHashSet<String/* topicName */>>
+ brokerIdIndexMap = new ConcurrentHashMap<>();
+
+ public BrokerTopicInfoView() {
+
+ }
+
+ // remove broker all topic info
+ public void rmvBrokerTopicInfo(int brokerId) {
+ TopicInfoView topicInfoView;
+ // remove pub info
+ ConcurrentHashSet<String> topicSet =
+ brokerIdIndexMap.remove(brokerId);
+ if (topicSet == null || topicSet.isEmpty()) {
+ return;
+ }
+ for (String topic : topicSet) {
+ if (topic == null) {
+ continue;
+ }
+ topicInfoView = topicConfInfoMap.get(topic);
+ if (topicInfoView == null
+ || topicInfoView.brokerTopicInfoMap.isEmpty()) {
+ continue;
+ }
+ topicInfoView.rmvBrokerTopicInfo(brokerId);
+ }
+ }
+
+ /**
+ * update broker's topicInfo configures
+ *
+ * @param brokerId broker id index
+ * @param topicInfoMap broker's topic configure info,
+ * if topicInfoMap is null, reserve current configure;
+ * if topicInfoMap is empty, clear current configure.
+ */
+ public void updBrokerTopicConfInfo(int brokerId,
+ Map<String, TopicInfo> topicInfoMap) {
+ if (topicInfoMap == null) {
+ return;
+ }
+ // get removed topic info
+ Set<String> delTopicSet = new HashSet<>();
+ ConcurrentHashSet<String> curTopicSet = brokerIdIndexMap.get(brokerId);
+ if (curTopicSet != null) {
+ for (String topic : curTopicSet) {
+ if (!topicInfoMap.containsKey(topic)) {
+ delTopicSet.add(topic);
+ }
+ }
+ }
+ rmvBrokerTopicInfo(brokerId, delTopicSet);
+ // add or update TopicInfo
+ repBrokerTopicInfo(brokerId, topicInfoMap);
+ }
+
+ /**
+ * Get the maximum number of broker distributions of topic
+ *
+ * @param topicSet need query topic set
+ */
+ public int getMaxTopicBrokerCnt(Set<String> topicSet) {
+ int maxCount = -1;
+ int tmpSize;
+ TopicInfoView topicInfoView;
+ if (topicSet == null || topicSet.isEmpty()) {
+ return maxCount;
+ }
+ for (String topic : topicSet) {
+ if (topic == null) {
+ continue;
+ }
+ topicInfoView = topicConfInfoMap.get(topic);
+ if (topicInfoView == null
+ || topicInfoView.brokerTopicInfoMap.isEmpty()) {
+ continue;
+ }
+ tmpSize = topicInfoView.curMapSize.get();
+ if (maxCount < tmpSize) {
+ maxCount = tmpSize;
+ }
+ }
+ return maxCount;
+ }
+
+ /**
+ * Gets the list of topic partitions whose subscribe status is enabled
+ *
+ * @param topicSet need query topic set
+ */
+ public List<Partition> getAcceptSubParts(Set<String> topicSet,
+ Set<Integer>
enableSubBrokerIdSet) {
+ List<Partition> partList = new ArrayList<>();
+ if (topicSet == null || topicSet.isEmpty()) {
+ return partList;
+ }
+ for (String topic : topicSet) {
+ partList.addAll(getAcceptSubParts(topic, enableSubBrokerIdSet));
+ }
+ return partList;
+ }
+
+ /**
+ * Gets the list of topic partitions whose subscribe status is enabled
+ *
+ * @param topic need query topic set
+ */
+ public List<Partition> getAcceptSubParts(String topic,
+ Set<Integer>
enableSubBrokerIdSet) {
+ TopicInfo topicInfo;
+ List<Partition> partList = new ArrayList<>();
+ if (topic == null) {
+ return partList;
+ }
+ TopicInfoView topicInfoView = topicConfInfoMap.get(topic);
+ if (topicInfoView == null
+ || topicInfoView.brokerTopicInfoMap.isEmpty()) {
+ return partList;
+ }
+ for (Map.Entry<Integer, TopicInfo> entry
+ : topicInfoView.brokerTopicInfoMap.entrySet()) {
+ if (entry.getKey() == null
+ || entry.getValue() == null
+ || !enableSubBrokerIdSet.contains(entry.getKey())) {
+ continue;
+ }
+ topicInfo = entry.getValue();
+ if (topicInfo.isAcceptSubscribe()) {
+ for (int j = 0; j < topicInfo.getTopicStoreNum(); j++) {
+ int baseValue = j * TBaseConstants.META_STORE_INS_BASE;
+ for (int i = 0; i < topicInfo.getPartitionNum(); i++) {
+ partList.add(new Partition(topicInfo.getBroker(),
+ topicInfo.getTopic(), baseValue + i));
+ }
+ }
+ }
+ }
+ return partList;
+ }
+
+ /**
+ * Gets the string map of topic partitions whose publish status is enabled
+ *
+ * @param topicSet need query topic set
+ */
+ public Map<String, String> getAcceptPubPartInfo(Set<String> topicSet,
+ Set<Integer>
enablePubBrokerIdSet) {
+ TopicInfo topicInfo;
+ TopicInfoView topicInfoView;
+ Map<String, String> topicPartStrMap = new HashMap<>();
+ Map<String, StringBuilder> topicPartBufferMap =
+ new HashMap<>();
+ for (String topic : topicSet) {
+ if (topic == null) {
+ continue;
+ }
+ topicInfoView = topicConfInfoMap.get(topic);
+ if (topicInfoView == null
+ || topicInfoView.brokerTopicInfoMap.isEmpty()) {
+ continue;
+ }
+ for (Map.Entry<Integer, TopicInfo> entry
+ : topicInfoView.brokerTopicInfoMap.entrySet()) {
+ if (entry.getKey() == null
+ || entry.getValue() == null
+ || !enablePubBrokerIdSet.contains(entry.getKey())) {
+ continue;
+ }
+ topicInfo = entry.getValue();
+ if (topicInfo.isAcceptPublish()) {
+ StringBuilder tmpValue = topicPartBufferMap.get(topic);
+ if (tmpValue == null) {
+ StringBuilder strBuffer =
+ new StringBuilder(512).append(topic)
+ .append(TokenConstants.SEGMENT_SEP)
+ .append(topicInfo.getSimpleValue());
+ topicPartBufferMap.put(topic, strBuffer);
+ } else {
+ tmpValue.append(TokenConstants.ARRAY_SEP)
+ .append(topicInfo.getSimpleValue());
+ }
+ }
+ }
+ }
+ for (Map.Entry<String, StringBuilder> entry :
topicPartBufferMap.entrySet()) {
+ if (entry.getValue() != null) {
+ topicPartStrMap.put(entry.getKey(),
entry.getValue().toString());
+ }
+ }
+ topicPartBufferMap.clear();
+ return topicPartStrMap;
+ }
+
+ /**
+ * Get the TopicInfo information of topic in broker
+ *
+ * @param brokerId need query broker
+ * @param topic need query topic
+ *
+ * @return null or topicInfo configure
+ */
+ public TopicInfo getBrokerPushedTopicInfo(int brokerId, String topic) {
+ TopicInfoView topicInfoView = topicConfInfoMap.get(topic);
+ if (topicInfoView == null) {
+ return null;
+ }
+ return topicInfoView.brokerTopicInfoMap.get(brokerId);
+ }
+
+ /**
+ * Get all TopicInfo information of broker
+ *
+ * @param brokerId need query broker
+ */
+ public List<TopicInfo> getBrokerPushedTopicInfo(int brokerId) {
+ TopicInfo topicInfo;
+ TopicInfoView topicInfoView;
+ List<TopicInfo> topicInfoList = new ArrayList<>();
+ ConcurrentHashSet<String> topicSet = brokerIdIndexMap.get(brokerId);
+ if (topicSet == null) {
+ return topicInfoList;
+ }
+ for (String topic : topicSet) {
+ if (topic == null) {
+ continue;
+ }
+ topicInfoView = topicConfInfoMap.get(topic);
+ if (topicInfoView == null
+ || topicInfoView.brokerTopicInfoMap.isEmpty()) {
+ continue;
+ }
+ topicInfo = topicInfoView.brokerTopicInfoMap.get(brokerId);
+ if (topicInfo == null) {
+ continue;
+ }
+ topicInfoList.add(topicInfo);
+ }
+ return topicInfoList;
+ }
+
+ // remove broker special topic info
+ private void rmvBrokerTopicInfo(int brokerId,
+ Set<String> delTopicSet) {
+ if (delTopicSet == null || delTopicSet.isEmpty()) {
+ return;
+ }
+ ConcurrentHashSet<String> topicSet =
+ brokerIdIndexMap.get(brokerId);
+ TopicInfoView topicInfoView;
+ if (topicSet == null || topicSet.isEmpty()) {
+ return;
+ }
+ for (String topic : delTopicSet) {
+ topicSet.remove(topic);
+ topicInfoView = topicConfInfoMap.get(topic);
+ if ((topicInfoView == null)
+ || topicInfoView.brokerTopicInfoMap.isEmpty()) {
+ continue;
+ }
+ topicInfoView.rmvBrokerTopicInfo(brokerId);
+ }
+ }
+
+ // add or update broker special topic info
+ private void repBrokerTopicInfo(int brokerId,
+ Map<String, TopicInfo> topicInfoMap) {
+ if (topicInfoMap == null || topicInfoMap.isEmpty()) {
+ return;
+ }
+ // add topic info
+ TopicInfoView newTopicInfoView;
+ TopicInfoView curTopicInfoView;
+ for (TopicInfo topicInfo : topicInfoMap.values()) {
+ if (topicInfo == null) {
+ continue;
+ }
+ curTopicInfoView = topicConfInfoMap.get(topicInfo.getTopic());
+ if (curTopicInfoView == null) {
+ newTopicInfoView = new TopicInfoView();
+ curTopicInfoView = topicConfInfoMap.putIfAbsent(
+ topicInfo.getTopic(), newTopicInfoView);
+ if (curTopicInfoView == null) {
+ curTopicInfoView = newTopicInfoView;
+ }
+ }
+ curTopicInfoView.addOrUpdBrokerTopicInfo(brokerId, topicInfo);
+ }
+ // add broker index
+ ConcurrentHashSet<String> curTopicSet = brokerIdIndexMap.get(brokerId);
+ if (curTopicSet == null) {
+ ConcurrentHashSet<String> newTopicSet = new ConcurrentHashSet<>();
+ curTopicSet = brokerIdIndexMap.putIfAbsent(brokerId, newTopicSet);
+ if (curTopicSet == null) {
+ curTopicSet = newTopicSet;
+ }
+ }
+ curTopicSet.addAll(topicInfoMap.keySet());
+ }
+
+ private static class TopicInfoView {
+ public AtomicInteger curMapSize = new AtomicInteger(0);
+ public AtomicLong topicChangeId = new AtomicLong(0);
+ public ConcurrentHashMap<Integer/* brokerId */, TopicInfo>
brokerTopicInfoMap =
+ new ConcurrentHashMap<Integer/* brokerId */, TopicInfo>();
+
+ public TopicInfoView() {
+
+ }
+
+ public boolean rmvBrokerTopicInfo(int brokerId) {
+ TopicInfo topicInfo = brokerTopicInfoMap.remove(brokerId);
+ if (topicInfo == null) {
+ return false;
+ }
+ curMapSize.decrementAndGet();
+ topicChangeId.set(System.currentTimeMillis());
+ return true;
+ }
+
+ public boolean addOrUpdBrokerTopicInfo(int brokerId, TopicInfo
topicInfo) {
+ if (topicInfo == null) {
+ return false;
+ }
+ TopicInfo newTopicInfo = topicInfo.clone();
+ TopicInfo preTopicInfo =
+ brokerTopicInfoMap.put(brokerId, newTopicInfo);
+ if (preTopicInfo == null) {
+ curMapSize.incrementAndGet();
+ }
+ topicChangeId.set(System.currentTimeMillis());
+ return true;
+ }
+ }
+}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/TargetValidResult.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/TargetValidResult.java
deleted file mode 100644
index 6a44f97..0000000
---
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/nodemanage/nodebroker/TargetValidResult.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.tubemq.server.master.nodemanage.nodebroker;
-
-
-public class TargetValidResult {
- public boolean result;
- public int errCode;
- public String errInfo;
-
- public TargetValidResult(boolean result, int errCode, final String
errInfo) {
- this.result = result;
- this.errCode = errCode;
- this.errInfo = errInfo;
- }
-
-}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/utils/BdbStoreSamplePrint.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/utils/BdbStoreSamplePrint.java
index a92acdc..7c5fb1b 100644
---
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/utils/BdbStoreSamplePrint.java
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/utils/BdbStoreSamplePrint.java
@@ -96,4 +96,13 @@ public class BdbStoreSamplePrint extends AbstractSamplePrint
{
}
}
+ @Override
+ public void printWarn(String err) {
+ //
+ }
+
+ @Override
+ public void printError(String err) {
+ //
+ }
}
diff --git
a/tubemq-server/src/main/java/org/apache/tubemq/server/master/utils/BrokerStatusSamplePrint.java
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/utils/BrokerStatusSamplePrint.java
new file mode 100644
index 0000000..b098eba
--- /dev/null
+++
b/tubemq-server/src/main/java/org/apache/tubemq/server/master/utils/BrokerStatusSamplePrint.java
@@ -0,0 +1,82 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.tubemq.server.master.utils;
+
+import org.apache.tubemq.corebase.utils.AbstractSamplePrint;
+import org.slf4j.Logger;
+
+
+public class BrokerStatusSamplePrint extends AbstractSamplePrint {
+ /**
+ * Log limit class
+ */
+ private final Logger logger;
+
+ public BrokerStatusSamplePrint(final Logger logger) {
+ super();
+ this.logger = logger;
+ }
+
+ public BrokerStatusSamplePrint(final Logger logger,
+ long sampleDetailDur, long sampleResetDur,
+ long maxDetailCount, long maxTotalCount) {
+ super(sampleDetailDur, sampleResetDur, maxDetailCount, maxTotalCount);
+ this.logger = logger;
+ }
+
+ @Override
+ public void printExceptionCaught(Throwable e) {
+ //
+ }
+
+ @Override
+ public void printExceptionCaught(Throwable e, String hostName, String
nodeName) {
+ //
+ }
+
+ @Override
+ public void printWarn(String err) {
+ printMsg(true, err);
+ }
+
+ @Override
+ public void printError(String err) {
+ printMsg(false, err);
+ }
+
+ private void printMsg(boolean isWarn, String err) {
+ final long now = System.currentTimeMillis();
+ final long diffTime = now - lastLogTime.get();
+ final long curPrintCnt = totalPrintCount.incrementAndGet();
+ if (curPrintCnt < maxTotalCount) {
+ if (diffTime < sampleDetailDur && curPrintCnt < maxDetailCount) {
+ if (isWarn) {
+ logger.warn(err);
+ } else {
+ logger.error(err);
+ }
+ }
+ }
+ if (diffTime > sampleResetDur) {
+ if (this.lastLogTime.compareAndSet(now - diffTime, now)) {
+ totalPrintCount.set(0);
+ }
+ }
+ }
+
+}
