This is an automated email from the ASF dual-hosted git repository.
dongeforever pushed a commit to branch 5.0.0-alpha-static-topic
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/5.0.0-alpha-static-topic by
this push:
new ad90cc1 Finish the slave sync logic for topic queue mapping
ad90cc1 is described below
commit ad90cc163c48e590a1aab16911f84fc704b72c97
Author: dongeforever <[email protected]>
AuthorDate: Wed Dec 8 20:38:31 2021 +0800
Finish the slave sync logic for topic queue mapping
---
.../apache/rocketmq/broker/out/BrokerOuterAPI.java | 4 ++--
.../broker/processor/AdminBrokerProcessor.java | 12 ++++++++++-
.../rocketmq/broker/slave/SlaveSynchronize.java | 15 +++++++++++---
.../TopicConfigAndMappingSerializeWrapper.java | 23 ++++++++++++++++++++++
4 files changed, 48 insertions(+), 6 deletions(-)
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
index e9ef46e..6a5b31e 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
@@ -350,7 +350,7 @@ public class BrokerOuterAPI {
return changedList;
}
- public TopicConfigSerializeWrapper getAllTopicConfig(
+ public TopicConfigAndMappingSerializeWrapper getAllTopicConfig(
final String addr) throws RemotingConnectException,
RemotingSendRequestException,
RemotingTimeoutException, InterruptedException, MQBrokerException {
RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
@@ -359,7 +359,7 @@ public class BrokerOuterAPI {
assert response != null;
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
- return TopicConfigSerializeWrapper.decode(response.getBody(),
TopicConfigSerializeWrapper.class);
+ return TopicConfigSerializeWrapper.decode(response.getBody(),
TopicConfigAndMappingSerializeWrapper.class);
}
default:
break;
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index e4f3697..2d39538 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -64,6 +64,7 @@ import
org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
import org.apache.rocketmq.common.protocol.body.QueryConsumeTimeSpanBody;
import org.apache.rocketmq.common.protocol.body.QueryCorrectionOffsetBody;
import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
+import
org.apache.rocketmq.common.protocol.body.TopicConfigAndMappingSerializeWrapper;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.body.UnlockBatchRequestBody;
import
org.apache.rocketmq.common.protocol.header.CloneGroupOffsetRequestHeader;
@@ -545,7 +546,16 @@ public class AdminBrokerProcessor extends
AsyncNettyRequestProcessor implements
// final GetAllTopicConfigResponseHeader responseHeader =
// (GetAllTopicConfigResponseHeader) response.readCustomHeader();
- String content =
this.brokerController.getTopicConfigManager().encode();
+ TopicConfigAndMappingSerializeWrapper
topicConfigAndMappingSerializeWrapper = new
TopicConfigAndMappingSerializeWrapper();
+
+
topicConfigAndMappingSerializeWrapper.setDataVersion(this.brokerController.getTopicConfigManager().getDataVersion());
+
topicConfigAndMappingSerializeWrapper.setTopicConfigTable(this.brokerController.getTopicConfigManager().getTopicConfigTable());
+
+
topicConfigAndMappingSerializeWrapper.setMappingDataVersion(this.brokerController.getTopicQueueMappingManager().getDataVersion());
+
topicConfigAndMappingSerializeWrapper.setTopicQueueMappingDetailMap(this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable());
+
+
+ String content = topicConfigAndMappingSerializeWrapper.toJson();
if (content != null && content.length() > 0) {
try {
response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET));
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
b/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
index 7b5e564..0029318 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
@@ -21,6 +21,7 @@ import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
+import
org.apache.rocketmq.common.protocol.body.TopicConfigAndMappingSerializeWrapper;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.protocol.body.ConsumerOffsetSerializeWrapper;
@@ -56,7 +57,7 @@ public class SlaveSynchronize {
String masterAddrBak = this.masterAddr;
if (masterAddrBak != null &&
!masterAddrBak.equals(brokerController.getBrokerAddr())) {
try {
- TopicConfigSerializeWrapper topicWrapper =
+ TopicConfigAndMappingSerializeWrapper topicWrapper =
this.brokerController.getBrokerOuterAPI().getAllTopicConfig(masterAddrBak);
if
(!this.brokerController.getTopicConfigManager().getDataVersion()
.equals(topicWrapper.getDataVersion())) {
@@ -67,9 +68,17 @@ public class SlaveSynchronize {
this.brokerController.getTopicConfigManager().getTopicConfigTable()
.putAll(topicWrapper.getTopicConfigTable());
this.brokerController.getTopicConfigManager().persist();
-
- log.info("Update slave topic config from master, {}",
masterAddrBak);
}
+ if (topicWrapper.getTopicQueueMappingDetailMap() != null
+ &&
!topicWrapper.getMappingDataVersion().equals(this.brokerController.getTopicQueueMappingManager().getDataVersion()))
{
+
this.brokerController.getTopicQueueMappingManager().getDataVersion()
+
.assignNewOne(topicWrapper.getMappingDataVersion());
+
this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable().clear();
+
this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable()
+
.putAll(topicWrapper.getTopicQueueMappingDetailMap());
+
this.brokerController.getTopicQueueMappingManager().persist();
+ }
+ log.info("Update slave topic config from master, {}",
masterAddrBak);
} catch (Exception e) {
log.error("SyncTopicConfig Exception, {}", masterAddrBak, e);
}
diff --git
a/common/src/main/java/org/apache/rocketmq/common/protocol/body/TopicConfigAndMappingSerializeWrapper.java
b/common/src/main/java/org/apache/rocketmq/common/protocol/body/TopicConfigAndMappingSerializeWrapper.java
index 47892e6..830a8a4 100644
---
a/common/src/main/java/org/apache/rocketmq/common/protocol/body/TopicConfigAndMappingSerializeWrapper.java
+++
b/common/src/main/java/org/apache/rocketmq/common/protocol/body/TopicConfigAndMappingSerializeWrapper.java
@@ -17,6 +17,8 @@
package org.apache.rocketmq.common.protocol.body;
+import org.apache.rocketmq.common.DataVersion;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
import java.util.Map;
@@ -25,6 +27,11 @@ import java.util.concurrent.ConcurrentHashMap;
public class TopicConfigAndMappingSerializeWrapper extends
TopicConfigSerializeWrapper {
private Map<String/* topic */, TopicQueueMappingInfo>
topicQueueMappingInfoMap = new ConcurrentHashMap<String,
TopicQueueMappingInfo>();
+ private Map<String/* topic */, TopicQueueMappingDetail>
topicQueueMappingDetailMap = new ConcurrentHashMap<String,
TopicQueueMappingDetail>();
+
+ private DataVersion mappingDataVersion = new DataVersion();
+
+
public Map<String, TopicQueueMappingInfo> getTopicQueueMappingInfoMap() {
return topicQueueMappingInfoMap;
}
@@ -33,6 +40,22 @@ public class TopicConfigAndMappingSerializeWrapper extends
TopicConfigSerializeW
this.topicQueueMappingInfoMap = topicQueueMappingInfoMap;
}
+ public Map<String, TopicQueueMappingDetail>
getTopicQueueMappingDetailMap() {
+ return topicQueueMappingDetailMap;
+ }
+
+ public void setTopicQueueMappingDetailMap(Map<String,
TopicQueueMappingDetail> topicQueueMappingDetailMap) {
+ this.topicQueueMappingDetailMap = topicQueueMappingDetailMap;
+ }
+
+ public DataVersion getMappingDataVersion() {
+ return mappingDataVersion;
+ }
+
+ public void setMappingDataVersion(DataVersion mappingDataVersion) {
+ this.mappingDataVersion = mappingDataVersion;
+ }
+
public static TopicConfigAndMappingSerializeWrapper
from(TopicConfigSerializeWrapper wrapper) {
if (wrapper instanceof TopicConfigAndMappingSerializeWrapper) {
return (TopicConfigAndMappingSerializeWrapper) wrapper;
