Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Summer of Code] Support switch role for broker #4272

Merged
merged 19 commits into from
May 16, 2022
Merged

[Summer of Code] Support switch role for broker #4272

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
22c424a
feature:
hzh0425 May 7, 2022
21648e0
feature:
hzh0425 May 7, 2022
31062dc
feature:
hzh0425 May 8, 2022
1f8b8f3
feature:
hzh0425 May 8, 2022
5668c3b
feature: add api message empty constructor
hzh0425 May 8, 2022
da3ecef
feature: move set from header to remotingRequest body
hzh0425 May 9, 2022
2377700
feature: modify autoSwitchHaClient's rpc protocol, add slaveId, slave…
hzh0425 May 10, 2022
04e9aa0
feature: review code
hzh0425 May 10, 2022
d4e2fd2
feature: review code
hzh0425 May 10, 2022
d1d336d
add some debug info
hzh0425 May 11, 2022
d373801
feature: let ha service get masterHaAdress after register to name-srv
hzh0425 May 11, 2022
78fbed5
feature: review code xxxx
hzh0425 May 11, 2022
e83ffae
feature: let controller return err remark
hzh0425 May 11, 2022
11e4ea6
style: review
hzh0425 May 11, 2022
947f7fc
style: review code
hzh0425 May 12, 2022
ede85b7
feature: add more integrationTest
hzh0425 May 13, 2022
153714b
style: review code
hzh0425 May 13, 2022
e758264
style: review code
hzh0425 May 13, 2022
c41aefc
fix: port already bind
hzh0425 May 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 43 additions & 21 deletions broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.apache.commons.io.FilenameUtils;
import org.apache.rocketmq.acl.AccessValidator;
import org.apache.rocketmq.broker.client.ClientHousekeepingService;
Expand All @@ -56,6 +55,7 @@
import org.apache.rocketmq.broker.filter.CommitLogDispatcherCalcBitMap;
import org.apache.rocketmq.broker.filter.ConsumerFilterManager;
import org.apache.rocketmq.broker.filtersrv.FilterServerManager;
import org.apache.rocketmq.broker.hacontroller.ReplicasManager;
import org.apache.rocketmq.broker.latency.BrokerFastFailure;
import org.apache.rocketmq.broker.latency.BrokerFixedThreadPoolExecutor;
import org.apache.rocketmq.broker.loadbalance.AssignmentManager;
Expand Down Expand Up @@ -114,6 +114,7 @@
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.common.protocol.NamespaceUtil;
import org.apache.rocketmq.common.protocol.RequestCode;
Expand All @@ -139,7 +140,6 @@
import org.apache.rocketmq.srvutil.FileWatchService;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.config.BrokerRole;
Expand Down Expand Up @@ -252,6 +252,7 @@ public class BrokerController {
protected volatile String minBrokerAddrInGroup = null;
private final Lock lock = new ReentrantLock();
protected final List<ScheduledFuture<?>> scheduledFutures = new ArrayList<>();
protected ReplicasManager replicasManager;

public BrokerController(
final BrokerConfig brokerConfig,
Expand Down Expand Up @@ -602,7 +603,7 @@ public void run() {
}
}, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);

if (!messageStoreConfig.isEnableDLegerCommitLog() && !messageStoreConfig.isDuplicationEnable()) {
if (!messageStoreConfig.isEnableDLegerCommitLog() && !messageStoreConfig.isDuplicationEnable() && !messageStoreConfig.isStartupControllerMode()) {
if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= HA_ADDRESS_MIN_LENGTH) {
this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
Expand Down Expand Up @@ -637,6 +638,10 @@ public void run() {
}, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
}
}

if (this.messageStoreConfig.isStartupControllerMode()) {
this.updateMasterHAServerAddrPeriodically = true;
}
}

protected void initializeScheduledTasks() {
Expand Down Expand Up @@ -706,7 +711,9 @@ public boolean initialize() throws CloneNotSupportedException {
MessageStorePluginContext context = new MessageStorePluginContext(this, messageStoreConfig, brokerStatsManager, messageArrivingListener);
this.messageStore = MessageStoreFactory.build(context, defaultMessageStore);
this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));

if (this.messageStoreConfig.isStartupControllerMode()) {
this.replicasManager = new ReplicasManager(this);
}
} catch (IOException e) {
result = false;
LOG.error("BrokerController#initialize: unexpected error occurs", e);
Expand Down Expand Up @@ -1355,6 +1362,10 @@ protected void startBasicService() throws Exception {
this.messageStore.start();
}

if (this.replicasManager != null) {
this.replicasManager.start();
}

if (remotingServerStartLatch != null) {
remotingServerStartLatch.await();
}
Expand Down Expand Up @@ -1441,7 +1452,7 @@ public void start() throws Exception {

startBasicService();

if (!isIsolated && !this.messageStoreConfig.isEnableDLegerCommitLog() && !this.messageStoreConfig.isDuplicationEnable()) {
if (!isIsolated && !this.messageStoreConfig.isEnableDLegerCommitLog() && !this.messageStoreConfig.isDuplicationEnable() && !this.messageStoreConfig.isStartupControllerMode()) {
changeSpecialServiceStatus(this.brokerConfig.getBrokerId() == MixAll.MASTER_ID);
this.registerBrokerAll(true, false, true);
}
Expand All @@ -1466,20 +1477,7 @@ public void run2() {
}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS));

if (this.brokerConfig.isEnableSlaveActingMaster()) {
scheduledFutures.add(this.brokerHeartbeatExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
@Override
public void run2() {
if (isIsolated) {
return;
}
try {
BrokerController.this.sendHeartbeat();
} catch (Exception e) {
BrokerController.LOG.error("sendHeartbeat Exception", e);
}

}
}, 1000, brokerConfig.getBrokerHeartbeatInterval(), TimeUnit.MILLISECONDS));
scheduleSendHeartbeat();

scheduledFutures.add(this.syncBrokerMemberGroupExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
@Override public void run2() {
Expand All @@ -1492,11 +1490,32 @@ public void run2() {
}, 1000, this.brokerConfig.getSyncBrokerMemberGroupPeriod(), TimeUnit.MILLISECONDS));
}

if (this.messageStoreConfig.isStartupControllerMode()) {
scheduleSendHeartbeat();
}

if (brokerConfig.isSkipPreOnline()) {
startServiceWithoutCondition();
}
}

protected void scheduleSendHeartbeat() {
scheduledFutures.add(this.brokerHeartbeatExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
@Override
public void run2() {
if (isIsolated) {
return;
}
try {
BrokerController.this.sendHeartbeat();
} catch (Exception e) {
BrokerController.LOG.error("sendHeartbeat Exception", e);
}

}
}, 1000, brokerConfig.getBrokerHeartbeatInterval(), TimeUnit.MILLISECONDS));
}

public synchronized void registerIncrementBrokerData(TopicConfig topicConfig, DataVersion dataVersion) {
this.registerIncrementBrokerData(Collections.singletonList(topicConfig), dataVersion);
}
Expand Down Expand Up @@ -1593,7 +1612,7 @@ protected void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
this.brokerConfig.getRegisterBrokerTimeoutMills(),
this.brokerConfig.isEnableSlaveActingMaster(),
this.brokerConfig.isCompressedRegister(),
this.brokerConfig.isEnableSlaveActingMaster() ? this.brokerConfig.getBrokerNotActiveTimeoutMillis() : null,
(this.brokerConfig.isEnableSlaveActingMaster() || this.messageStoreConfig.isStartupControllerMode()) ? this.brokerConfig.getBrokerNotActiveTimeoutMillis() : null,
this.getBrokerIdentity());

handleRegisterBrokerResult(registerBrokerResultList, checkOrderConfig);
Expand Down Expand Up @@ -1657,7 +1676,6 @@ protected void handleRegisterBrokerResult(List<RegisterBrokerResult> registerBro
}

this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());

if (checkOrderConfig) {
this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
}
Expand Down Expand Up @@ -2131,6 +2149,10 @@ public BrokerIdentity getBrokerIdentity() {
}
}

public ReplicasManager getReplicasManager() {
return replicasManager;
}

public boolean isIsolated() {
return this.isIsolated;
}
Expand Down
Loading