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 10 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
61 changes: 40 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 @@ -706,7 +707,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, this.messageStore);
}
} catch (IOException e) {
result = false;
LOG.error("BrokerController#initialize: unexpected error occurs", e);
Expand Down Expand Up @@ -1355,6 +1358,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 +1448,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 +1473,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 +1486,32 @@ public void run2() {
}, 1000, this.brokerConfig.getSyncBrokerMemberGroupPeriod(), TimeUnit.MILLISECONDS));
}

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

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

private 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 +1608,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 @@ -1652,12 +1667,12 @@ protected void handleRegisterBrokerResult(List<RegisterBrokerResult> registerBro
boolean checkOrderConfig) {
for (RegisterBrokerResult registerBrokerResult : registerBrokerResultList) {
if (registerBrokerResult != null) {
System.out.println("Handle broker:" + getBrokerAddr() + " registered result, master address:" + registerBrokerResult.getMasterAddr());
if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) {
this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
}

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

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

public ReplicasManager getReplicasManager() {
return replicasManager;
}

public boolean isIsolated() {
return this.isIsolated;
}
Expand Down
176 changes: 176 additions & 0 deletions broker/src/main/java/org/apache/rocketmq/broker/hacontroller/HaControllerProxy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,176 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.rocketmq.broker.hacontroller;

import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.body.SyncStateSet;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetMetaDataResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.GetReplicaInfoResponseHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerRequestHeader;
import org.apache.rocketmq.common.protocol.header.namesrv.controller.RegisterBrokerResponseHeader;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;

import static org.apache.rocketmq.common.protocol.ResponseCode.CONTROLLER_NOT_LEADER;
import static org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode.SUCCESS;

/**
* The proxy of controller api.
*/
public class HaControllerProxy {
private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
public static final int RPC_TIME_OUT = 3000;
private final RemotingClient remotingClient;
private final List<String> controllerAddresses;
private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ControllerProxy_"));
private volatile String controllerLeaderAddress = "";

public HaControllerProxy(final NettyClientConfig nettyClientConfig, final List<String> controllerAddresses) {
this.remotingClient = new NettyRemotingClient(nettyClientConfig);
hzh0425 marked this conversation as resolved.
Show resolved Hide resolved
this.controllerAddresses = controllerAddresses;
}

public boolean start() {
this.remotingClient.start();
// Get controller metadata first.
int tryTimes = 0;
while (tryTimes < 3) {
boolean flag = updateControllerMetadata();
if (flag) {
this.executorService.scheduleAtFixedRate(this::updateControllerMetadata, 0, 2, TimeUnit.SECONDS);
return true;
}
tryTimes ++;
}
LOGGER.error("Failed to init controller metadata, maybe the controllers in {} is not available", this.controllerAddresses);
return false;
}

public void shutdown() {
this.remotingClient.shutdown();
this.executorService.shutdown();
}

/**
* Update controller metadata(leaderAddress)
*/
private boolean updateControllerMetadata() {
for (final String address : this.controllerAddresses) {
final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_METADATA_INFO, null);
try {
final RemotingCommand response = this.remotingClient.invokeSync(address, request, RPC_TIME_OUT);
if (response.getCode() == SUCCESS) {
final GetMetaDataResponseHeader responseHeader = response.decodeCommandCustomHeader(GetMetaDataResponseHeader.class);
if (responseHeader != null && responseHeader.isLeader()) {
// Because the controller is served externally with the help of name-srv
this.controllerLeaderAddress = address;
LOGGER.info("Change controller leader address to {}", this.controllerAddresses);
return true;
}
}
} catch (final Exception e) {
LOGGER.error("Error happen when pull controller metadata", e);
return false;
}
}
return false;
}

/**
* Alter syncStateSet
*/
public SyncStateSet alterSyncStateSet(String brokerName,
final String masterAddress, final int masterEpoch,
final Set<String> newSyncStateSet, final int syncStateSetEpoch) throws Exception {

final AlterSyncStateSetRequestHeader requestHeader = new AlterSyncStateSetRequestHeader(brokerName, masterAddress, masterEpoch);
final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET, requestHeader);
request.setBody(new SyncStateSet(newSyncStateSet, syncStateSetEpoch).encode());
final RemotingCommand response = this.remotingClient.invokeSync(this.controllerLeaderAddress, request, RPC_TIME_OUT);
assert response != null;
switch (response.getCode()) {
case SUCCESS: {
assert response.getBody() != null;
return RemotingSerializable.decode(response.getBody(), SyncStateSet.class);
}
case CONTROLLER_NOT_LEADER: {
throw new MQBrokerException(response.getCode(), "Controller leader was changed");
}
}
hzh0425 marked this conversation as resolved.
Show resolved Hide resolved
throw new MQBrokerException(response.getCode(), response.getRemark());
}

/**
* Register broker to controller
*/
public RegisterBrokerResponseHeader registerBroker(final String clusterName, final String brokerName,
final String address, final String haAddress) throws Exception {

final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader(clusterName, brokerName, address, haAddress);
final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_REGISTER_BROKER, requestHeader);
final RemotingCommand response = this.remotingClient.invokeSync(this.controllerLeaderAddress, request, RPC_TIME_OUT);
assert response != null;
switch (response.getCode()) {
case SUCCESS: {
return response.decodeCommandCustomHeader(RegisterBrokerResponseHeader.class);
}
case CONTROLLER_NOT_LEADER: {
throw new MQBrokerException(response.getCode(), "Controller leader was changed");
}
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}

/**
* Get broker replica info
*/
public Pair<GetReplicaInfoResponseHeader, SyncStateSet> getReplicaInfo(final String brokerName) throws Exception {
final GetReplicaInfoRequestHeader requestHeader = new GetReplicaInfoRequestHeader(brokerName);
final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_REPLICA_INFO, requestHeader);
final RemotingCommand response = this.remotingClient.invokeSync(this.controllerLeaderAddress, request, RPC_TIME_OUT);
assert response != null;
switch (response.getCode()) {
case SUCCESS: {
final GetReplicaInfoResponseHeader header = response.decodeCommandCustomHeader(GetReplicaInfoResponseHeader.class);
assert response.getBody() != null;
final SyncStateSet stateSet = RemotingSerializable.decode(response.getBody(), SyncStateSet.class);
return new Pair<>(header, stateSet);
}
case CONTROLLER_NOT_LEADER: {
throw new MQBrokerException(response.getCode(), "Controller leader was changed");
}
}
throw new MQBrokerException(response.getCode(), response.getRemark());
}
}
Loading