opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-28 20:09:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #7300] jRaft-Controller Implemention (#7301)

Browse Source 
* jRaft-Controller Implemention

* fix bazel build

* reformat code

* remove fury dependence

* fix bazel build

* resolve conflict

* clear code

* Optimize code style

* Optimize code style

* Optimize code style

* revert style only change
fix code style
rollback an unexpected modification.

* Update Producer.java

* chore: move raft startup to start method

* chore: use jraftconfig to collect all configs about jraft

* fix: fix wrong store path because init use string constant

* fix: fix CONTROLLER_NOT_LEADER error in follower

* chore: seperate jraft and controller log

* chore: fix conflict with develop

* chore: add comment to clear the filter logic

* feat: triggerElectMaster will retry when failed

* feat: when controller all restart, we use a timestamp to trace the first heartbeat, avoid to elect again

* fix: implements Serializable to enable snapshot serialize

* fix: use for loop to simple the elect retry

* chore: update jraft version

* chore: opt import

---------

Co-authored-by: leizhiyuan <leizhiyuan@gmail.com>
This commit is contained in:
yulangz
2024-01-29 20:45:20 +08:00
committed by GitHub
parent 178421386f
commit 8df53dfde8
57 changed files with 2485 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/BUILD.bazel
+3 -1
View File
@@ -51,6 +51,8 @@ java_library(
"@maven//:io_opentelemetry_opentelemetry_exporter_logging",
"@maven//:io_opentelemetry_opentelemetry_exporter_logging_otlp",
"@maven//:org_slf4j_jul_to_slf4j",
"@maven//:com_alipay_sofa_jraft_core",
"@maven//:com_alipay_sofa_hessian",
],
)
@@ -69,7 +71,7 @@ java_library(
"@maven//:org_apache_commons_commons_lang3",
"@maven//:io_netty_netty_all",
"@maven//:com_google_guava_guava",
"@maven//:com_alibaba_fastjson",
"@maven//:com_alibaba_fastjson",
],
resources = glob(["src/test/resources/certs/*.pem"]) + glob(["src/test/resources/certs/*.key"])
)
controller/pom.xml
+10 -1
View File
@@ -15,7 +15,8 @@
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>rocketmq-all</artifactId>
<groupId>org.apache.rocketmq</groupId>
@@ -62,5 +63,13 @@
<groupId>org.slf4j</groupId>
<artifactId>jul-to-slf4j</artifactId>
</dependency>
<dependency>
<groupId>com.alipay.sofa</groupId>
<artifactId>jraft-core</artifactId>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java-util</artifactId>
</dependency>
</dependencies>
</project>
controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java
+17 -1
View File
@@ -17,17 +17,32 @@
package org.apache.rocketmq.controller;
import io.netty.channel.Channel;
import java.util.Map;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.controller.helper.BrokerLifecycleListener;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
import org.apache.rocketmq.controller.impl.heartbeat.RaftBrokerHeartBeatManager;
import java.util.Map;
public interface BrokerHeartbeatManager {
public static final long DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 10;
public static BrokerHeartbeatManager newBrokerHeartbeatManager(ControllerConfig controllerConfig) {
if (controllerConfig.getControllerType().equals(ControllerConfig.JRAFT_CONTROLLER)) {
return new RaftBrokerHeartBeatManager(controllerConfig);
} else {
return new DefaultBrokerHeartbeatManager(controllerConfig);
}
}
/**
* initialize the resources
*
* @return
*/
void initialize();
/**
* Broker new heartbeat.
*/
@@ -67,6 +82,7 @@ public interface BrokerHeartbeatManager {
/**
* Count the number of active brokers in each broker-set of each cluster
*
* @return active brokers count
*/
Map<String/*cluster*/, Map<String/*broker-set*/, Integer/*active broker num*/>> getActiveBrokersNum();
controller/src/main/java/org/apache/rocketmq/controller/ControllerManager.java
+64 -28
View File
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.controller;
import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
@@ -26,17 +27,16 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.controller.elect.impl.DefaultElectPolicy;
import org.apache.rocketmq.controller.impl.DLedgerController;
import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
import org.apache.rocketmq.controller.impl.JRaftController;
import org.apache.rocketmq.controller.impl.heartbeat.RaftBrokerHeartBeatManager;
import org.apache.rocketmq.controller.metrics.ControllerMetricsManager;
import org.apache.rocketmq.controller.processor.ControllerRequestProcessor;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -68,12 +68,10 @@ public class ControllerManager {
private final Configuration configuration;
private final RemotingClient remotingClient;
private Controller controller;
private BrokerHeartbeatManager heartbeatManager;
private final BrokerHeartbeatManager heartbeatManager;
private ExecutorService controllerRequestExecutor;
private BlockingQueue<Runnable> controllerRequestThreadPoolQueue;
private NotifyService notifyService;
private final NotifyService notifyService;
private ControllerMetricsManager controllerMetricsManager;
public ControllerManager(ControllerConfig controllerConfig, NettyServerConfig nettyServerConfig,
@@ -85,7 +83,7 @@ public class ControllerManager {
this.configuration = new Configuration(log, this.controllerConfig, this.nettyServerConfig);
this.configuration.setStorePathFromConfig(this.controllerConfig, "configStorePath");
this.remotingClient = new NettyRemotingClient(nettyClientConfig);
this.heartbeatManager = new DefaultBrokerHeartbeatManager(this.controllerConfig);
this.heartbeatManager = BrokerHeartbeatManager.newBrokerHeartbeatManager(controllerConfig);
this.notifyService = new NotifyService();
}
@@ -100,15 +98,31 @@ public class ControllerManager {
new ThreadFactoryImpl("ControllerRequestExecutorThread_"));
this.notifyService.initialize();
if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerPeers())) {
throw new IllegalArgumentException("Attribute value controllerDLegerPeers of ControllerConfig is null or empty");
if (controllerConfig.getControllerType().equals(ControllerConfig.JRAFT_CONTROLLER)) {
if (StringUtils.isEmpty(this.controllerConfig.getJraftConfig().getjRaftInitConf())) {
throw new IllegalArgumentException("Attribute value jRaftInitConf of ControllerConfig is null or empty");
}
if (StringUtils.isEmpty(this.controllerConfig.getJraftConfig().getjRaftServerId())) {
throw new IllegalArgumentException("Attribute value jRaftServerId of ControllerConfig is null or empty");
}
try {
this.controller = new JRaftController(controllerConfig, this.brokerHousekeepingService);
((RaftBrokerHeartBeatManager) this.heartbeatManager).setController((JRaftController) this.controller);
} catch (IOException e) {
throw new RuntimeException(e);
}
} else {
if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerPeers())) {
throw new IllegalArgumentException("Attribute value controllerDLegerPeers of ControllerConfig is null or empty");
}
if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerSelfId())) {
throw new IllegalArgumentException("Attribute value controllerDLegerSelfId of ControllerConfig is null or empty");
}
this.controller = new DLedgerController(this.controllerConfig, this.heartbeatManager::isBrokerActive,
this.nettyServerConfig, this.nettyClientConfig, this.brokerHousekeepingService,
new DefaultElectPolicy(this.heartbeatManager::isBrokerActive, this.heartbeatManager::getBrokerLiveInfo));
}
if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerSelfId())) {
throw new IllegalArgumentException("Attribute value controllerDLegerSelfId of ControllerConfig is null or empty");
}
this.controller = new DLedgerController(this.controllerConfig, this.heartbeatManager::isBrokerActive,
this.nettyServerConfig, this.nettyClientConfig, this.brokerHousekeepingService,
new DefaultElectPolicy(this.heartbeatManager::isBrokerActive, this.heartbeatManager::getBrokerLiveInfo));
// Initialize the basic resources
this.heartbeatManager.initialize();
@@ -126,10 +140,12 @@ public class ControllerManager {
* something else.
*
* @param clusterName The cluster name of this inactive broker
* @param brokerName The inactive broker name
* @param brokerId The inactive broker id, null means that the election forced to be triggered
* @param brokerName The inactive broker name
* @param brokerId The inactive broker id, null means that the election forced to be triggered
*/
private void onBrokerInactive(String clusterName, String brokerName, Long brokerId) {
log.info("Controller Manager received broker inactive event, clusterName: {}, brokerName: {}, brokerId: {}",
clusterName, brokerName, brokerId);
if (controller.isLeaderState()) {
if (brokerId == null) {
// Means that force triggering election for this broker-set
@@ -156,22 +172,39 @@ public class ControllerManager {
}
}
private void triggerElectMaster(String brokerName) {
private CompletableFuture<Boolean> triggerElectMaster0(String brokerName) {
final CompletableFuture<RemotingCommand> electMasterFuture = controller.electMaster(ElectMasterRequestHeader.ofControllerTrigger(brokerName));
electMasterFuture.whenCompleteAsync((electMasterResponse, err) -> {
if (err != null || electMasterResponse == null) {
return electMasterFuture.handleAsync((electMasterResponse, err) -> {
if (err != null || electMasterResponse == null || electMasterResponse.getCode() != ResponseCode.SUCCESS) {
log.error("Failed to trigger elect-master in broker-set: {}", brokerName, err);
return;
return false;
}
if (electMasterResponse.getCode() == ResponseCode.SUCCESS) {
log.info("Elect a new master in broker-set: {} done, result: {}", brokerName, electMasterResponse);
if (controllerConfig.isNotifyBrokerRoleChanged()) {
notifyBrokerRoleChanged(RoleChangeNotifyEntry.convert(electMasterResponse));
}
return true;
}
//default is false
return false;
});
}
private void triggerElectMaster(String brokerName) {
int maxRetryCount = controllerConfig.getElectMasterMaxRetryCount();
for (int i = 0; i < maxRetryCount; i++) {
try {
Boolean electResult = triggerElectMaster0(brokerName).get(3, TimeUnit.SECONDS);
if (electResult) {
return;
}
} catch (Exception e) {
log.warn("Failed to trigger elect-master in broker-set: {}, retryCount: {}", brokerName, i, e);
}
}
}
/**
* Notify master and all slaves for a broker that the master role changed.
*/
@@ -188,20 +221,21 @@ public class ControllerManager {
// Inform all active brokers
final Map<Long, String> brokerAddrs = memberGroup.getBrokerAddrs();
brokerAddrs.entrySet().stream().filter(x -> this.heartbeatManager.isBrokerActive(clusterName, brokerName, x.getKey()))
.forEach(x -> this.notifyService.notifyBroker(x.getValue(), entry));
.forEach(x -> this.notifyService.notifyBroker(x.getValue(), entry));
}
}
/**
* Notify broker that there are roles-changing in controller
*
* @param brokerAddr target broker's address to notify
* @param entry role change entry
* @param entry role change entry
*/
public void doNotifyBrokerRoleChanged(final String brokerAddr, final RoleChangeNotifyEntry entry) {
if (StringUtils.isNoneEmpty(brokerAddr)) {
log.info("Try notify broker {} that role changed, RoleChangeNotifyEntry:{}", brokerAddr, entry);
final NotifyBrokerRoleChangedRequestHeader requestHeader = new NotifyBrokerRoleChangedRequestHeader(entry.getMasterAddress(), entry.getMasterBrokerId(),
entry.getMasterEpoch(), entry.getSyncStateSetEpoch());
entry.getMasterEpoch(), entry.getSyncStateSetEpoch());
final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.NOTIFY_BROKER_ROLE_CHANGED, requestHeader);
request.setBody(new SyncStateSet(entry.getSyncStateSet(), entry.getSyncStateSetEpoch()).encode());
try {
@@ -214,7 +248,7 @@ public class ControllerManager {
public void registerProcessor() {
final ControllerRequestProcessor controllerRequestProcessor = new ControllerRequestProcessor(this);
final RemotingServer controllerRemotingServer = this.controller.getRemotingServer();
RemotingServer controllerRemotingServer = this.controller.getRemotingServer();
assert controllerRemotingServer != null;
controllerRemotingServer.registerProcessor(RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET, controllerRequestProcessor, this.controllerRequestExecutor);
controllerRemotingServer.registerProcessor(RequestCode.CONTROLLER_ELECT_MASTER, controllerRequestProcessor, this.controllerRequestExecutor);
@@ -231,8 +265,8 @@ public class ControllerManager {
}
public void start() {
this.heartbeatManager.start();
this.controller.startup();
this.heartbeatManager.start();
this.remotingClient.start();
}
@@ -335,7 +369,9 @@ public class ControllerManager {
@Override
public boolean equals(Object obj) {
if (this == obj) return true;
if (this == obj) {
return true;
}
if (!(obj instanceof NotifyTask)) {
return false;
}
controller/src/main/java/org/apache/rocketmq/controller/ControllerStartup.java
+5
View File
@@ -28,6 +28,7 @@ import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.JraftConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -74,6 +75,8 @@ public class ControllerStartup {
}
final ControllerConfig controllerConfig = new ControllerConfig();
final JraftConfig jraftConfig = new JraftConfig();
controllerConfig.setJraftConfig(jraftConfig);
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
final NettyClientConfig nettyClientConfig = new NettyClientConfig();
nettyServerConfig.setListenPort(19876);
@@ -85,6 +88,7 @@ public class ControllerStartup {
properties = new Properties();
properties.load(in);
MixAll.properties2Object(properties, controllerConfig);
MixAll.properties2Object(properties, jraftConfig);
MixAll.properties2Object(properties, nettyServerConfig);
MixAll.properties2Object(properties, nettyClientConfig);
@@ -96,6 +100,7 @@ public class ControllerStartup {
if (commandLine.hasOption('p')) {
Logger console = LoggerFactory.getLogger(LoggerName.CONTROLLER_CONSOLE_NAME);
MixAll.printObjectProperties(console, controllerConfig);
MixAll.printObjectProperties(console, jraftConfig);
MixAll.printObjectProperties(console, nettyServerConfig);
MixAll.printObjectProperties(console, nettyClientConfig);
System.exit(0);
controller/src/main/java/org/apache/rocketmq/controller/elect/ElectPolicy.java
+2 -2
View File
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.controller.elect;
import java.util.Set;
public interface ElectPolicy {
@@ -32,6 +31,7 @@ public interface ElectPolicy {
* @param brokerId broker id(can be used as prefer or assigned in some elect policy)
* @return new master's broker id
*/
Long elect(String clusterName, String brokerName, Set<Long> syncStateBrokers, Set<Long> allReplicaBrokers, Long oldMaster, Long brokerId);
Long elect(String clusterName, String brokerName, Set<Long> syncStateBrokers, Set<Long> allReplicaBrokers,
Long oldMaster, Long brokerId);
}
controller/src/main/java/org/apache/rocketmq/controller/elect/impl/DefaultElectPolicy.java
+5 -5
View File
@@ -39,7 +39,7 @@ public class DefaultElectPolicy implements ElectPolicy {
private final Comparator<BrokerLiveInfo> comparator = (o1, o2) -> {
if (o1.getEpoch() == o2.getEpoch()) {
return o1.getMaxOffset() == o2.getMaxOffset() ? o1.getElectionPriority() - o2.getElectionPriority() :
(int) (o2.getMaxOffset() - o1.getMaxOffset());
(int) (o2.getMaxOffset() - o1.getMaxOffset());
} else {
return o2.getEpoch() - o1.getEpoch();
}
@@ -70,7 +70,8 @@ public class DefaultElectPolicy implements ElectPolicy {
* @return master elected by our own policy
*/
@Override
public Long elect(String clusterName, String brokerName, Set<Long> syncStateBrokers, Set<Long> allReplicaBrokers, Long oldMaster, Long preferBrokerId) {
public Long elect(String clusterName, String brokerName, Set<Long> syncStateBrokers, Set<Long> allReplicaBrokers,
Long oldMaster, Long preferBrokerId) {
Long newMaster = null;
// try to elect in syncStateBrokers
if (syncStateBrokers != null) {
@@ -87,8 +88,8 @@ public class DefaultElectPolicy implements ElectPolicy {
return newMaster;
}
private Long tryElect(String clusterName, String brokerName, Set<Long> brokers, Long oldMaster, Long preferBrokerId) {
private Long tryElect(String clusterName, String brokerName, Set<Long> brokers, Long oldMaster,
Long preferBrokerId) {
if (this.validPredicate != null) {
brokers = brokers.stream().filter(brokerAddr -> this.validPredicate.check(clusterName, brokerName, brokerAddr)).collect(Collectors.toSet());
}
@@ -118,7 +119,6 @@ public class DefaultElectPolicy implements ElectPolicy {
}
public void setBrokerLiveInfoGetter(BrokerLiveInfoGetter brokerLiveInfoGetter) {
this.brokerLiveInfoGetter = brokerLiveInfoGetter;
}
controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerController.java
+6 -6
View File
@@ -66,11 +66,11 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerRequestHeader;
@@ -99,14 +99,14 @@ public class DLedgerController implements Controller {
private ScheduledFuture scanInactiveMasterFuture;
private List<BrokerLifecycleListener> brokerLifecycleListeners;
private final List<BrokerLifecycleListener> brokerLifecycleListeners;
// Usr for checking whether the broker is alive
private BrokerValidPredicate brokerAlivePredicate;
// use for elect a master
private ElectPolicy electPolicy;
private AtomicBoolean isScheduling = new AtomicBoolean(false);
private final AtomicBoolean isScheduling = new AtomicBoolean(false);
public DLedgerController(final ControllerConfig config, final BrokerValidPredicate brokerAlivePredicate) {
this(config, brokerAlivePredicate, null, null, null, null);
@@ -555,8 +555,8 @@ public class DLedgerController implements Controller {
if (DLedgerController.this.scanInactiveMasterFuture == null) {
long scanInactiveMasterInterval = DLedgerController.this.controllerConfig.getScanInactiveMasterInterval();
DLedgerController.this.scanInactiveMasterFuture =
DLedgerController.this.scanInactiveMasterService.scheduleAtFixedRate(DLedgerController.this::scanInactiveMasterAndTriggerReelect,
scanInactiveMasterInterval, scanInactiveMasterInterval, TimeUnit.MILLISECONDS);
DLedgerController.this.scanInactiveMasterService.scheduleAtFixedRate(DLedgerController.this::scanInactiveMasterAndTriggerReelect,
scanInactiveMasterInterval, scanInactiveMasterInterval, TimeUnit.MILLISECONDS);
}
break;
}
controller/src/main/java/org/apache/rocketmq/controller/impl/DLedgerControllerStateMachine.java
+2 -1
View File
@@ -21,7 +21,6 @@ import io.openmessaging.storage.dledger.snapshot.SnapshotReader;
import io.openmessaging.storage.dledger.snapshot.SnapshotWriter;
import io.openmessaging.storage.dledger.statemachine.CommittedEntryIterator;
import io.openmessaging.storage.dledger.statemachine.StateMachine;
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.controller.impl.event.EventMessage;
import org.apache.rocketmq.controller.impl.event.EventSerializer;
@@ -29,6 +28,8 @@ import org.apache.rocketmq.controller.impl.manager.ReplicasInfoManager;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import java.util.concurrent.CompletableFuture;
/**
* The state machine implementation of the dledger controller
*/
controller/src/main/java/org/apache/rocketmq/controller/impl/JRaftController.java
+279
View File
@@ -0,0 +1,279 @@
/*
* 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.controller.impl;
import com.alipay.sofa.jraft.Node;
import com.alipay.sofa.jraft.RaftGroupService;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.conf.Configuration;
import com.alipay.sofa.jraft.entity.NodeId;
import com.alipay.sofa.jraft.entity.PeerId;
import com.alipay.sofa.jraft.entity.Task;
import com.alipay.sofa.jraft.option.NodeOptions;
import org.apache.commons.io.FileUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.controller.Controller;
import org.apache.rocketmq.controller.helper.BrokerLifecycleListener;
import org.apache.rocketmq.controller.impl.closure.ControllerClosure;
import org.apache.rocketmq.controller.impl.task.BrokerCloseChannelRequest;
import org.apache.rocketmq.controller.impl.task.CheckNotActiveBrokerRequest;
import org.apache.rocketmq.controller.impl.task.GetBrokerLiveInfoRequest;
import org.apache.rocketmq.controller.impl.task.GetSyncStateDataRequest;
import org.apache.rocketmq.controller.impl.task.RaftBrokerHeartBeatEventRequest;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.netty.NettyRemotingServer;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerRequestHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
public class JRaftController implements Controller {
private static final Logger log = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private final RaftGroupService raftGroupService;
private Node node;
private final JRaftControllerStateMachine stateMachine;
private final ControllerConfig controllerConfig;
private final List<BrokerLifecycleListener> brokerLifecycleListeners;
private final Map<PeerId/* jRaft peerId */, String/* Controller RPC Server Addr */> peerIdToAddr;
private final NettyRemotingServer remotingServer;
public JRaftController(ControllerConfig controllerConfig,
final ChannelEventListener channelEventListener) throws IOException {
this.controllerConfig = controllerConfig;
this.brokerLifecycleListeners = new ArrayList<>();
final NodeOptions nodeOptions = new NodeOptions();
nodeOptions.setElectionTimeoutMs(controllerConfig.getJraftConfig().getjRaftElectionTimeoutMs());
nodeOptions.setSnapshotIntervalSecs(controllerConfig.getJraftConfig().getjRaftSnapshotIntervalSecs());
final PeerId serverId = new PeerId();
if (!serverId.parse(controllerConfig.getJraftConfig().getjRaftServerId())) {
throw new IllegalArgumentException("Fail to parse serverId:" + controllerConfig.getJraftConfig().getjRaftServerId());
}
final Configuration initConf = new Configuration();
if (!initConf.parse(controllerConfig.getJraftConfig().getjRaftInitConf())) {
throw new IllegalArgumentException("Fail to parse initConf:" + controllerConfig.getJraftConfig().getjRaftInitConf());
}
nodeOptions.setInitialConf(initConf);
FileUtils.forceMkdir(new File(controllerConfig.getControllerStorePath()));
nodeOptions.setLogUri(controllerConfig.getControllerStorePath() + File.separator + "log");
nodeOptions.setRaftMetaUri(controllerConfig.getControllerStorePath() + File.separator + "raft_meta");
nodeOptions.setSnapshotUri(controllerConfig.getControllerStorePath() + File.separator + "snapshot");
this.stateMachine = new JRaftControllerStateMachine(controllerConfig, new NodeId(controllerConfig.getJraftConfig().getjRaftGroupId(), serverId));
this.stateMachine.registerOnLeaderStart(this::onLeaderStart);
this.stateMachine.registerOnLeaderStop(this::onLeaderStop);
nodeOptions.setFsm(this.stateMachine);
this.raftGroupService = new RaftGroupService(controllerConfig.getJraftConfig().getjRaftGroupId(), serverId, nodeOptions);
this.peerIdToAddr = new HashMap<>();
initPeerIdMap();
NettyServerConfig nettyServerConfig = new NettyServerConfig();
nettyServerConfig.setListenPort(Integer.parseInt(this.peerIdToAddr.get(serverId).split(":")[1]));
remotingServer = new NettyRemotingServer(nettyServerConfig, channelEventListener);
}
private void initPeerIdMap() {
String[] peers = this.controllerConfig.getJraftConfig().getjRaftInitConf().split(",");
String[] rpcAddrs = this.controllerConfig.getJraftConfig().getjRaftControllerRPCAddr().split(",");
for (int i = 0; i < peers.length; i++) {
PeerId peerId = new PeerId();
if (!peerId.parse(peers[i])) {
throw new IllegalArgumentException("Fail to parse peerId:" + peers[i]);
}
this.peerIdToAddr.put(peerId, rpcAddrs[i]);
}
}
@Override
public void startup() {
this.remotingServer.start();
this.node = this.raftGroupService.start();
log.info("Controller {} started.", node.getNodeId());
}
@Override
public void shutdown() {
this.stopScheduling();
this.raftGroupService.shutdown();
this.remotingServer.shutdown();
log.info("Controller {} stopped.", node.getNodeId());
}
@Override
public void startScheduling() {
}
@Override
public void stopScheduling() {
}
@Override
public boolean isLeaderState() {
return node.isLeader();
}
private <T extends CommandCustomHeader> CompletableFuture<RemotingCommand> applyToJRaft(RemotingCommand request) {
if (!isLeaderState()) {
final RemotingCommand command = RemotingCommand.createResponseCommand(ResponseCode.CONTROLLER_NOT_LEADER, "The controller is not in leader state");
final CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
future.complete(command);
log.warn("Apply to none leader controller, controller state is {}", node.getNodeState());
return future;
}
ControllerClosure closure = new ControllerClosure(request);
Task task = closure.taskWithThisClosure();
if (task != null) {
node.apply(task);
return closure.getFuture();
} else {
log.error("Apply task failed, task is null.");
return CompletableFuture.completedFuture(RemotingCommand.createResponseCommand(ResponseCode.CONTROLLER_JRAFT_INTERNAL_ERROR, "Apply task failed, Please see the server log."));
}
}
@Override
public CompletableFuture<RemotingCommand> alterSyncStateSet(AlterSyncStateSetRequestHeader request,
SyncStateSet syncStateSet) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET, request);
requestCommand.setBody(syncStateSet.encode());
return applyToJRaft(requestCommand);
}
@Override
public CompletableFuture<RemotingCommand> electMaster(ElectMasterRequestHeader request) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_ELECT_MASTER, request);
return applyToJRaft(requestCommand);
}
@Override
public CompletableFuture<RemotingCommand> getNextBrokerId(GetNextBrokerIdRequestHeader request) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_NEXT_BROKER_ID, request);
return applyToJRaft(requestCommand);
}
@Override
public CompletableFuture<RemotingCommand> applyBrokerId(ApplyBrokerIdRequestHeader request) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_APPLY_BROKER_ID, request);
return applyToJRaft(requestCommand);
}
@Override
public CompletableFuture<RemotingCommand> registerBroker(RegisterBrokerToControllerRequestHeader request) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_REGISTER_BROKER, request);
return applyToJRaft(requestCommand);
}
@Override
public CompletableFuture<RemotingCommand> getReplicaInfo(GetReplicaInfoRequestHeader request) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_REPLICA_INFO, request);
return applyToJRaft(requestCommand);
}
@Override
public CompletableFuture<RemotingCommand> getSyncStateData(List<String> brokerNames) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_SYNC_STATE_DATA, new GetSyncStateDataRequest());
requestCommand.setBody(RemotingSerializable.encode(brokerNames));
return applyToJRaft(requestCommand);
}
@Override
public CompletableFuture<RemotingCommand> cleanBrokerData(CleanControllerBrokerDataRequestHeader requestHeader) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CLEAN_BROKER_DATA, requestHeader);
return applyToJRaft(requestCommand);
}
@Override
public void registerBrokerLifecycleListener(BrokerLifecycleListener listener) {
this.brokerLifecycleListeners.add(listener);
}
@Override
public RemotingCommand getControllerMetadata() {
List<PeerId> peers = node.getOptions().getInitialConf().getPeers();
final StringBuilder sb = new StringBuilder();
for (PeerId peer : peers) {
sb.append(peerIdToAddr.get(peer)).append(";");
}
return RemotingCommand.createResponseCommandWithHeader(ResponseCode.SUCCESS, new GetMetaDataResponseHeader(
node.getGroupId(),
node.getLeaderId() == null ? "" : node.getLeaderId().toString(),
this.peerIdToAddr.get(node.getLeaderId()),
node.isLeader(),
sb.toString()
));
}
@Override
public RemotingServer getRemotingServer() {
return remotingServer;
}
public void onLeaderStart(long term) {
log.info("Controller start leadership, term: {}.", term);
}
public void onLeaderStop(Status status) {
log.info("Controller {} stop leadership, status: {}.", node.getNodeId(), status);
this.stopScheduling();
}
public CompletableFuture<RemotingCommand> getBrokerLiveInfo(GetBrokerLiveInfoRequest requestHeader) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_LIVE_INFO_REQUEST, requestHeader);
return applyToJRaft(requestCommand);
}
public CompletableFuture<RemotingCommand> onBrokerHeartBeat(RaftBrokerHeartBeatEventRequest requestHeader) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.RAFT_BROKER_HEART_BEAT_EVENT_REQUEST, requestHeader);
return applyToJRaft(requestCommand);
}
public CompletableFuture<RemotingCommand> onBrokerCloseChannel(BrokerCloseChannelRequest requestHeader) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.BROKER_CLOSE_CHANNEL_REQUEST, requestHeader);
return applyToJRaft(requestCommand);
}
public CompletableFuture<RemotingCommand> checkNotActiveBroker(CheckNotActiveBrokerRequest requestHeader) {
final RemotingCommand requestCommand = RemotingCommand.createRequestCommand(RequestCode.CHECK_NOT_ACTIVE_BROKER_REQUEST, requestHeader);
return applyToJRaft(requestCommand);
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/JRaftControllerStateMachine.java
+331
View File
@@ -0,0 +1,331 @@
/*
* 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.controller.impl;
import com.alipay.sofa.jraft.Closure;
import com.alipay.sofa.jraft.Iterator;
import com.alipay.sofa.jraft.StateMachine;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.conf.Configuration;
import com.alipay.sofa.jraft.entity.LeaderChangeContext;
import com.alipay.sofa.jraft.entity.NodeId;
import com.alipay.sofa.jraft.error.RaftError;
import com.alipay.sofa.jraft.error.RaftException;
import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader;
import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter;
import com.alipay.sofa.jraft.util.Utils;
import io.opentelemetry.api.common.AttributesBuilder;
import org.apache.commons.io.FileUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.controller.elect.impl.DefaultElectPolicy;
import org.apache.rocketmq.controller.impl.closure.ControllerClosure;
import org.apache.rocketmq.controller.impl.event.ControllerResult;
import org.apache.rocketmq.controller.impl.manager.RaftReplicasInfoManager;
import org.apache.rocketmq.controller.impl.task.BrokerCloseChannelRequest;
import org.apache.rocketmq.controller.impl.task.CheckNotActiveBrokerRequest;
import org.apache.rocketmq.controller.impl.task.GetBrokerLiveInfoRequest;
import org.apache.rocketmq.controller.impl.task.GetSyncStateDataRequest;
import org.apache.rocketmq.controller.impl.task.RaftBrokerHeartBeatEventRequest;
import org.apache.rocketmq.controller.metrics.ControllerMetricsConstant;
import org.apache.rocketmq.controller.metrics.ControllerMetricsManager;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerRequestHeader;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_BROKER_SET;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_CLUSTER_NAME;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_ELECTION_RESULT;
public class JRaftControllerStateMachine implements StateMachine {
private static final Logger log = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private final List<Consumer<Long>> onLeaderStartCallbacks;
private final List<Consumer<Status>> onLeaderStopCallbacks;
private final RaftReplicasInfoManager replicasInfoManager;
private final NodeId nodeId;
public JRaftControllerStateMachine(ControllerConfig controllerConfig, NodeId nodeId) {
this.replicasInfoManager = new RaftReplicasInfoManager(controllerConfig);
this.nodeId = nodeId;
this.onLeaderStartCallbacks = new ArrayList<>();
this.onLeaderStopCallbacks = new ArrayList<>();
}
@Override
public void onApply(Iterator iter) {
while (iter.hasNext()) {
byte[] data = iter.getData().array();
ControllerClosure controllerClosure = (ControllerClosure) iter.done();
processEvent(controllerClosure, data, iter.getTerm(), iter.getIndex());
iter.next();
}
}
private void processEvent(ControllerClosure controllerClosure, byte[] data, long term, long index) {
RemotingCommand request;
ControllerResult<?> result;
try {
if (controllerClosure != null) {
request = controllerClosure.getRequestEvent();
} else {
request = RemotingCommand.decode(Arrays.copyOfRange(data, 4, data.length));
}
log.info("process event: term {}, index {}, request code {}", term, index, request.getCode());
switch (request.getCode()) {
case RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET:
AlterSyncStateSetRequestHeader requestHeader = (AlterSyncStateSetRequestHeader) request.decodeCommandCustomHeader(AlterSyncStateSetRequestHeader.class);
SyncStateSet syncStateSet = RemotingSerializable.decode(request.getBody(), SyncStateSet.class);
result = alterSyncStateSet(requestHeader, syncStateSet);
break;
case RequestCode.CONTROLLER_ELECT_MASTER:
ElectMasterRequestHeader electMasterRequestHeader = (ElectMasterRequestHeader) request.decodeCommandCustomHeader(ElectMasterRequestHeader.class);
result = electMaster(electMasterRequestHeader);
break;
case RequestCode.CONTROLLER_GET_NEXT_BROKER_ID:
GetNextBrokerIdRequestHeader getNextBrokerIdRequestHeader = (GetNextBrokerIdRequestHeader) request.decodeCommandCustomHeader(GetNextBrokerIdRequestHeader.class);
result = getNextBrokerId(getNextBrokerIdRequestHeader);
break;
case RequestCode.CONTROLLER_APPLY_BROKER_ID:
ApplyBrokerIdRequestHeader applyBrokerIdRequestHeader = (ApplyBrokerIdRequestHeader) request.decodeCommandCustomHeader(ApplyBrokerIdRequestHeader.class);
result = applyBrokerId(applyBrokerIdRequestHeader);
break;
case RequestCode.CONTROLLER_REGISTER_BROKER:
RegisterBrokerToControllerRequestHeader registerBrokerToControllerRequestHeader = (RegisterBrokerToControllerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerToControllerRequestHeader.class);
result = registerBroker(registerBrokerToControllerRequestHeader);
break;
case RequestCode.CONTROLLER_GET_REPLICA_INFO:
GetReplicaInfoRequestHeader getReplicaInfoRequestHeader = (GetReplicaInfoRequestHeader) request.decodeCommandCustomHeader(GetReplicaInfoRequestHeader.class);
result = getReplicaInfo(getReplicaInfoRequestHeader);
break;
case RequestCode.CONTROLLER_GET_SYNC_STATE_DATA:
List<String> brokerNames = RemotingSerializable.decode(request.getBody(), List.class);
GetSyncStateDataRequest getSyncStateDataRequest = (GetSyncStateDataRequest) request.decodeCommandCustomHeader(GetSyncStateDataRequest.class);
result = getSyncStateData(brokerNames, getSyncStateDataRequest.getInvokeTime());
break;
case RequestCode.CLEAN_BROKER_DATA:
CleanControllerBrokerDataRequestHeader cleanBrokerDataRequestHeader = (CleanControllerBrokerDataRequestHeader) request.decodeCommandCustomHeader(CleanControllerBrokerDataRequestHeader.class);
result = cleanBrokerData(cleanBrokerDataRequestHeader);
break;
case RequestCode.GET_BROKER_LIVE_INFO_REQUEST:
GetBrokerLiveInfoRequest getBrokerLiveInfoRequest = (GetBrokerLiveInfoRequest) request.decodeCommandCustomHeader(GetBrokerLiveInfoRequest.class);
result = replicasInfoManager.getBrokerLiveInfo(getBrokerLiveInfoRequest);
break;
case RequestCode.RAFT_BROKER_HEART_BEAT_EVENT_REQUEST:
RaftBrokerHeartBeatEventRequest brokerHeartbeatRequestHeader = (RaftBrokerHeartBeatEventRequest) request.decodeCommandCustomHeader(RaftBrokerHeartBeatEventRequest.class);
result = replicasInfoManager.onBrokerHeartBeat(brokerHeartbeatRequestHeader);
break;
case RequestCode.BROKER_CLOSE_CHANNEL_REQUEST:
BrokerCloseChannelRequest brokerCloseChannelRequest = (BrokerCloseChannelRequest) request.decodeCommandCustomHeader(BrokerCloseChannelRequest.class);
result = replicasInfoManager.onBrokerCloseChannel(brokerCloseChannelRequest);
break;
case RequestCode.CHECK_NOT_ACTIVE_BROKER_REQUEST:
CheckNotActiveBrokerRequest checkNotActiveBrokerRequest = (CheckNotActiveBrokerRequest) request.decodeCommandCustomHeader(CheckNotActiveBrokerRequest.class);
result = replicasInfoManager.checkNotActiveBroker(checkNotActiveBrokerRequest);
break;
default:
throw new RemotingCommandException("Unknown request code: " + request.getCode());
}
result.getEvents().forEach(replicasInfoManager::applyEvent);
} catch (RemotingCommandException e) {
log.error("Fail to process event", e);
if (controllerClosure != null) {
controllerClosure.run(new Status(RaftError.EINTERNAL, e.getMessage()));
}
return;
}
log.info("process event: term {}, index {}, request code {} success with result {}", term, index, request.getCode(), result.toString());
if (controllerClosure != null) {
controllerClosure.setControllerResult(result);
controllerClosure.run(Status.OK());
}
}
private ControllerResult<AlterSyncStateSetResponseHeader> alterSyncStateSet(
AlterSyncStateSetRequestHeader requestHeader, SyncStateSet syncStateSet) {
return replicasInfoManager.alterSyncStateSet(requestHeader, syncStateSet, new RaftReplicasInfoManager.BrokerValidPredicateWithInvokeTime(requestHeader.getInvokeTime(), this.replicasInfoManager));
}
private ControllerResult<ElectMasterResponseHeader> electMaster(ElectMasterRequestHeader request) {
ControllerResult<ElectMasterResponseHeader> electResult = this.replicasInfoManager.electMaster(request, new DefaultElectPolicy(
(clusterName, brokerName, brokerId) -> replicasInfoManager.isBrokerActive(clusterName, brokerName, brokerId, request.getInvokeTime()),
replicasInfoManager::getBrokerLiveInfo
));
log.info("elect master, request :{}, result: {}", request.toString(), electResult.toString());
AttributesBuilder attributesBuilder = ControllerMetricsManager.newAttributesBuilder()
.put(LABEL_CLUSTER_NAME, request.getClusterName())
.put(LABEL_BROKER_SET, request.getBrokerName());
switch (electResult.getResponseCode()) {
case ResponseCode.SUCCESS:
ControllerMetricsManager.electionTotal.add(1,
attributesBuilder.put(LABEL_ELECTION_RESULT, ControllerMetricsConstant.ElectionResult.NEW_MASTER_ELECTED.getLowerCaseName()).build());
break;
case ResponseCode.CONTROLLER_MASTER_STILL_EXIST:
ControllerMetricsManager.electionTotal.add(1,
attributesBuilder.put(LABEL_ELECTION_RESULT, ControllerMetricsConstant.ElectionResult.KEEP_CURRENT_MASTER.getLowerCaseName()).build());
break;
case ResponseCode.CONTROLLER_MASTER_NOT_AVAILABLE:
case ResponseCode.CONTROLLER_ELECT_MASTER_FAILED:
ControllerMetricsManager.electionTotal.add(1,
attributesBuilder.put(LABEL_ELECTION_RESULT, ControllerMetricsConstant.ElectionResult.NO_MASTER_ELECTED.getLowerCaseName()).build());
break;
default:
break;
}
return electResult;
}
private ControllerResult<GetNextBrokerIdResponseHeader> getNextBrokerId(
GetNextBrokerIdRequestHeader requestHeader) {
return replicasInfoManager.getNextBrokerId(requestHeader);
}
private ControllerResult<ApplyBrokerIdResponseHeader> applyBrokerId(ApplyBrokerIdRequestHeader requestHeader) {
return replicasInfoManager.applyBrokerId(requestHeader);
}
private ControllerResult<?> registerBroker(RegisterBrokerToControllerRequestHeader request) {
return replicasInfoManager.registerBroker(request, new RaftReplicasInfoManager.BrokerValidPredicateWithInvokeTime(request.getInvokeTime(), this.replicasInfoManager));
}
private ControllerResult<GetReplicaInfoResponseHeader> getReplicaInfo(GetReplicaInfoRequestHeader request) {
return replicasInfoManager.getReplicaInfo(request);
}
private ControllerResult<Void> getSyncStateData(List<String> brokerNames, long invokeTile) {
return replicasInfoManager.getSyncStateData(brokerNames, new RaftReplicasInfoManager.BrokerValidPredicateWithInvokeTime(invokeTile, this.replicasInfoManager));
}
private ControllerResult<Void> cleanBrokerData(CleanControllerBrokerDataRequestHeader requestHeader) {
return replicasInfoManager.cleanBrokerData(requestHeader, new RaftReplicasInfoManager.BrokerValidPredicateWithInvokeTime(requestHeader.getInvokeTime(), this.replicasInfoManager));
}
@Override
public void onShutdown() {
log.info("StateMachine {} node {} onShutdown", getClass().getName(), nodeId.toString());
}
@Override
public void onSnapshotSave(SnapshotWriter writer, Closure done) {
byte[] data;
try {
data = this.replicasInfoManager.serialize();
} catch (Throwable e) {
done.run(new Status(RaftError.EIO, "Fail to serialize replicasInfoManager state machine data"));
return;
}
Utils.runInThread(() -> {
try {
FileUtils.writeByteArrayToFile(new File(writer.getPath() + File.separator + "data"), data);
if (writer.addFile("data")) {
log.info("Save snapshot, path={}", writer.getPath());
done.run(Status.OK());
} else {
throw new IOException("Fail to add file to writer");
}
} catch (IOException e) {
log.error("Fail to save snapshot", e);
done.run(new Status(RaftError.EIO, "Fail to save snapshot"));
}
});
}
@Override
public boolean onSnapshotLoad(SnapshotReader reader) {
if (reader.getFileMeta("data") == null) {
log.error("Fail to find data file in {}", reader.getPath());
return false;
}
try {
byte[] data = FileUtils.readFileToByteArray(new File(reader.getPath() + File.separator + "data"));
this.replicasInfoManager.deserializeFrom(data);
log.info("Load snapshot from {}", reader.getPath());
return true;
} catch (Throwable e) {
log.error("Fail to load snapshot from {}", reader.getPath(), e);
return false;
}
}
@Override
public void onLeaderStart(long term) {
for (Consumer<Long> callback : onLeaderStartCallbacks) {
callback.accept(term);
}
log.info("node {} Start Leader, term={}", nodeId.toString(), term);
}
@Override
public void onLeaderStop(Status status) {
for (Consumer<Status> callback : onLeaderStopCallbacks) {
callback.accept(status);
}
log.info("node {} Stop Leader, status={}", nodeId.toString(), status);
}
public void registerOnLeaderStart(Consumer<Long> callback) {
onLeaderStartCallbacks.add(callback);
}
public void registerOnLeaderStop(Consumer<Status> callback) {
onLeaderStopCallbacks.add(callback);
}
@Override
public void onError(RaftException e) {
log.error("Encountered an error={} on StateMachine {}, node {}, raft may stop working since some error occurs, you should figure out the cause and repair or remove this node.", e.getStatus(), this.getClass().getName(), nodeId.toString(), e);
}
@Override
public void onConfigurationCommitted(Configuration conf) {
log.info("Configuration committed, conf={}", conf);
}
@Override
public void onStopFollowing(LeaderChangeContext ctx) {
log.info("Stop following, ctx={}", ctx);
}
@Override
public void onStartFollowing(LeaderChangeContext ctx) {
log.info("Start following, ctx={}", ctx);
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/closure/ControllerClosure.java
+83
View File
@@ -0,0 +1,83 @@
/*
* 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.controller.impl.closure;
import com.alipay.sofa.jraft.Closure;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.entity.Task;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.controller.impl.event.ControllerResult;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import java.util.concurrent.CompletableFuture;
public class ControllerClosure implements Closure {
private static final Logger log = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private final RemotingCommand requestEvent;
private final CompletableFuture<RemotingCommand> future;
private ControllerResult<?> controllerResult;
private Task task;
public ControllerClosure(RemotingCommand requestEvent) {
this.requestEvent = requestEvent;
this.future = new CompletableFuture<>();
this.task = null;
}
public CompletableFuture<RemotingCommand> getFuture() {
return future;
}
public void setControllerResult(ControllerResult<?> controllerResult) {
this.controllerResult = controllerResult;
}
@Override
public void run(Status status) {
if (status.isOk()) {
final RemotingCommand response = RemotingCommand.createResponseCommandWithHeader(controllerResult.getResponseCode(), (CommandCustomHeader) controllerResult.getResponse());
if (controllerResult.getBody() != null) {
response.setBody(controllerResult.getBody());
}
if (controllerResult.getRemark() != null) {
response.setRemark(controllerResult.getRemark());
}
future.complete(response);
} else {
log.error("Failed to append to jRaft node, error is: {}.", status);
future.complete(RemotingCommand.createResponseCommand(ResponseCode.CONTROLLER_JRAFT_INTERNAL_ERROR, status.getErrorMsg()));
}
}
public Task taskWithThisClosure() {
if (task != null) {
return task;
}
task = new Task();
task.setDone(this);
task.setData(requestEvent.encode());
return task;
}
public RemotingCommand getRequestEvent() {
return requestEvent;
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/event/ApplyBrokerIdEvent.java
+8 -7
View File
@@ -29,7 +29,8 @@ public class ApplyBrokerIdEvent implements EventMessage {
private final long newBrokerId;
public ApplyBrokerIdEvent(String clusterName, String brokerName, String brokerAddress, long newBrokerId, String registerCheckCode) {
public ApplyBrokerIdEvent(String clusterName, String brokerName, String brokerAddress, long newBrokerId,
String registerCheckCode) {
this.clusterName = clusterName;
this.brokerName = brokerName;
this.brokerAddress = brokerAddress;
@@ -65,11 +66,11 @@ public class ApplyBrokerIdEvent implements EventMessage {
@Override
public String toString() {
return "ApplyBrokerIdEvent{" +
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerAddress='" + brokerAddress + '\'' +
", registerCheckCode='" + registerCheckCode + '\'' +
", newBrokerId=" + newBrokerId +
'}';
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerAddress='" + brokerAddress + '\'' +
", registerCheckCode='" + registerCheckCode + '\'' +
", newBrokerId=" + newBrokerId +
'}';
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/event/CleanBrokerDataEvent.java
+3 -3
View File
@@ -57,8 +57,8 @@ public class CleanBrokerDataEvent implements EventMessage {
@Override
public String toString() {
return "CleanBrokerDataEvent{" +
"brokerName='" + brokerName + '\'' +
", brokerIdSetToClean=" + brokerIdSetToClean +
'}';
"brokerName='" + brokerName + '\'' +
", brokerIdSetToClean=" + brokerIdSetToClean +
'}';
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/event/ElectMasterEvent.java
+4 -4
View File
@@ -60,9 +60,9 @@ public class ElectMasterEvent implements EventMessage {
@Override
public String toString() {
return "ElectMasterEvent{" +
"newMasterElected=" + newMasterElected +
", brokerName='" + brokerName + '\'' +
", newMasterBrokerId=" + newMasterBrokerId +
'}';
"newMasterElected=" + newMasterElected +
", brokerName='" + brokerName + '\'' +
", newMasterBrokerId=" + newMasterBrokerId +
'}';
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/event/ListEventSerializer.java
+128
View File
@@ -0,0 +1,128 @@
/*
* 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.controller.impl.event;
import org.apache.commons.lang3.SerializationException;
import org.apache.rocketmq.common.utils.FastJsonSerializer;
import org.apache.rocketmq.common.utils.Serializer;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import java.io.ByteArrayOutputStream;
import java.util.ArrayList;
import java.util.List;
public class ListEventSerializer {
private ListEventSerializer() {
}
private static final Serializer SERIALIZER = new FastJsonSerializer();
private static void putShort(byte[] memory, int index, int value) {
memory[index] = (byte) (value >>> 8);
memory[index + 1] = (byte) value;
}
private static void putShort(ByteArrayOutputStream outputStream, int value) {
outputStream.write((byte) (value >>> 8));
outputStream.write((byte) value);
}
private static short getShort(byte[] memory, int index) {
return (short) (memory[index] << 8 | memory[index + 1] & 0xFF);
}
private static void putInt(byte[] memory, int index, int value) {
memory[index] = (byte) (value >>> 24);
memory[index + 1] = (byte) (value >>> 16);
memory[index + 2] = (byte) (value >>> 8);
memory[index + 3] = (byte) value;
}
private static void putInt(ByteArrayOutputStream outputStream, int value) {
outputStream.write((byte) (value >>> 24));
outputStream.write((byte) (value >>> 16));
outputStream.write((byte) (value >>> 8));
outputStream.write((byte) value);
}
private static int getInt(byte[] memory, int index) {
return memory[index] << 24 | (memory[index + 1] & 0xFF) << 16 | (memory[index + 2] & 0xFF) << 8 | memory[index + 3] & 0xFF;
}
public static byte[] serialize(List<EventMessage> message, Logger log) throws SerializationException {
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
for (EventMessage eventMessage : message) {
final short eventType = eventMessage.getEventType().getId();
final byte[] data = SERIALIZER.serialize(eventMessage);
if (data != null && data.length > 0) {
putShort(outputStream, eventType);
putInt(outputStream, data.length);
outputStream.write(data, 0, data.length);
} else {
log.error("serialize event message error, event: {}, this event will be discard", eventMessage);
}
}
return outputStream.toByteArray();
}
public static List<EventMessage> deserialize(byte[] bytes, Logger log) throws SerializationException {
List<EventMessage> eventMessages = new ArrayList<>();
if (bytes == null || bytes.length <= 6) {
return eventMessages;
}
int index = 0;
while (index < bytes.length) {
final short eventId = getShort(bytes, index);
index += 2;
final int dataLength = getInt(bytes, index);
index += 4;
if (dataLength > 0) {
final byte[] data = new byte[dataLength];
System.arraycopy(bytes, index, data, 0, dataLength);
final EventType eventType = EventType.from(eventId);
if (eventType != null) {
switch (eventType) {
case ALTER_SYNC_STATE_SET_EVENT:
eventMessages.add(SERIALIZER.deserialize(data, AlterSyncStateSetEvent.class));
break;
case APPLY_BROKER_ID_EVENT:
eventMessages.add(SERIALIZER.deserialize(data, ApplyBrokerIdEvent.class));
break;
case ELECT_MASTER_EVENT:
eventMessages.add(SERIALIZER.deserialize(data, ElectMasterEvent.class));
break;
case CLEAN_BROKER_DATA_EVENT:
eventMessages.add(SERIALIZER.deserialize(data, CleanBrokerDataEvent.class));
break;
case UPDATE_BROKER_ADDRESS:
eventMessages.add(SERIALIZER.deserialize(data, UpdateBrokerAddressEvent.class));
break;
default:
log.error("deserialize event message error, event id: {}, data: {}", eventId, data);
break;
}
} else {
log.error("deserialize event message error, event id: {}, data: {}", eventId, data);
}
index += dataLength;
} else {
log.error("deserialize event message error, event id: {}, data length: {}", eventId, dataLength);
}
}
return eventMessages;
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/event/UpdateBrokerAddressEvent.java
+5 -5
View File
@@ -53,11 +53,11 @@ public class UpdateBrokerAddressEvent implements EventMessage {
@Override
public String toString() {
return "UpdateBrokerAddressEvent{" +
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerAddress='" + brokerAddress + '\'' +
", brokerId=" + brokerId +
'}';
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerAddress='" + brokerAddress + '\'' +
", brokerId=" + brokerId +
'}';
}
@Override
controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/BrokerIdentityInfo.java
+11 -6
View File
@@ -16,9 +16,14 @@
*/
package org.apache.rocketmq.controller.impl.heartbeat;
import java.io.Serializable;
import org.apache.rocketmq.common.UtilAll;
import java.util.Objects;
public class BrokerIdentityInfo {
public class BrokerIdentityInfo implements Serializable {
private static final long serialVersionUID = 883597359635995567L;
private final String clusterName;
private final String brokerName;
@@ -44,7 +49,7 @@ public class BrokerIdentityInfo {
}
public boolean isEmpty() {
return clusterName.isEmpty() && brokerName.isEmpty() && brokerId == null;
return UtilAll.isBlank(clusterName) && UtilAll.isBlank(brokerName) && brokerId == null;
}
@Override
@@ -71,9 +76,9 @@ public class BrokerIdentityInfo {
@Override
public String toString() {
return "BrokerIdentityInfo{" +
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerId=" + brokerId +
'}';
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerId=" + brokerId +
'}';
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/BrokerLiveInfo.java
+5 -2
View File
@@ -17,8 +17,10 @@
package org.apache.rocketmq.controller.impl.heartbeat;
import io.netty.channel.Channel;
import java.io.Serializable;
public class BrokerLiveInfo {
public class BrokerLiveInfo implements Serializable {
private static final long serialVersionUID = 3612173344946510993L;
private final String brokerName;
private String brokerAddr;
@@ -45,7 +47,8 @@ public class BrokerLiveInfo {
}
public BrokerLiveInfo(String brokerName, String brokerAddr, long brokerId, long lastUpdateTimestamp,
long heartbeatTimeoutMillis, Channel channel, int epoch, long maxOffset, Integer electionPriority, long confirmOffset) {
long heartbeatTimeoutMillis, Channel channel, int epoch, long maxOffset, Integer electionPriority,
long confirmOffset) {
this.brokerName = brokerName;
this.brokerAddr = brokerAddr;
this.brokerId = brokerId;
controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java
+1 -1
View File
@@ -40,7 +40,7 @@ import org.apache.rocketmq.remoting.common.RemotingHelper;
public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
private static final Logger log = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private static final long DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 10;
private ScheduledExecutorService scheduledService;
private ExecutorService executor;
controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/RaftBrokerHeartBeatManager.java
+278
View File
@@ -0,0 +1,278 @@
/*
* 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.controller.impl.heartbeat;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import io.netty.channel.Channel;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.controller.BrokerHeartbeatManager;
import org.apache.rocketmq.controller.helper.BrokerLifecycleListener;
import org.apache.rocketmq.controller.impl.JRaftController;
import org.apache.rocketmq.controller.impl.task.BrokerCloseChannelRequest;
import org.apache.rocketmq.controller.impl.task.CheckNotActiveBrokerRequest;
import org.apache.rocketmq.controller.impl.task.GetBrokerLiveInfoRequest;
import org.apache.rocketmq.controller.impl.task.GetBrokerLiveInfoResponse;
import org.apache.rocketmq.controller.impl.task.RaftBrokerHeartBeatEventRequest;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class RaftBrokerHeartBeatManager implements BrokerHeartbeatManager {
private static final Logger log = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private JRaftController controller;
private final List<BrokerLifecycleListener> brokerLifecycleListeners = new ArrayList<>();
private final ScheduledExecutorService scheduledService;
private final ExecutorService executor;
private final ControllerConfig controllerConfig;
private final Map<Channel, BrokerIdentityInfo> brokerChannelIdentityInfoMap = new HashMap<>();
// resolve the scene
// when controller all down and startup again, we wait for some time to avoid electing a new leader,which is not necessary
private long firstReceivedHeartbeatTime = -1;
public RaftBrokerHeartBeatManager(ControllerConfig controllerConfig) {
this.scheduledService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("RaftBrokerHeartbeatManager_scheduledService_"));
this.executor = Executors.newFixedThreadPool(2, new ThreadFactoryImpl("RaftBrokerHeartbeatManager_executorService_"));
this.controllerConfig = controllerConfig;
}
public void setController(JRaftController controller) {
this.controller = controller;
}
@Override
public void initialize() {
}
@Override
public void start() {
this.scheduledService.scheduleAtFixedRate(this::scanNotActiveBroker, 2000, this.controllerConfig.getScanNotActiveBrokerInterval(), TimeUnit.MILLISECONDS);
}
@Override
public void shutdown() {
this.scheduledService.shutdown();
this.executor.shutdown();
}
@Override
public void registerBrokerLifecycleListener(BrokerLifecycleListener listener) {
brokerLifecycleListeners.add(listener);
}
@Override
public void onBrokerHeartbeat(String clusterName, String brokerName, String brokerAddr, Long brokerId,
Long timeoutMillis, Channel channel, Integer epoch, Long maxOffset, Long confirmOffset,
Integer electionPriority) {
if (firstReceivedHeartbeatTime == -1) {
firstReceivedHeartbeatTime = System.currentTimeMillis();
}
BrokerIdentityInfo brokerIdentityInfo = new BrokerIdentityInfo(clusterName, brokerName, brokerId);
int realEpoch = Optional.ofNullable(epoch).orElse(-1);
long realBrokerId = Optional.ofNullable(brokerId).orElse(-1L);
long realMaxOffset = Optional.ofNullable(maxOffset).orElse(-1L);
long realConfirmOffset = Optional.ofNullable(confirmOffset).orElse(-1L);
long realTimeoutMillis = Optional.ofNullable(timeoutMillis).orElse(DEFAULT_BROKER_CHANNEL_EXPIRED_TIME);
int realElectionPriority = Optional.ofNullable(electionPriority).orElse(Integer.MAX_VALUE);
BrokerLiveInfo liveInfo = new BrokerLiveInfo(brokerName,
brokerAddr,
realBrokerId,
System.currentTimeMillis(),
realTimeoutMillis,
null,
realEpoch,
realMaxOffset,
realElectionPriority,
realConfirmOffset);
log.info("broker {} heart beat", brokerIdentityInfo);
RaftBrokerHeartBeatEventRequest requestHeader = new RaftBrokerHeartBeatEventRequest(brokerIdentityInfo, liveInfo);
CompletableFuture<RemotingCommand> future = controller.onBrokerHeartBeat(requestHeader);
try {
RemotingCommand remotingCommand = future.get(5, java.util.concurrent.TimeUnit.SECONDS);
if (remotingCommand.getCode() != ResponseCode.SUCCESS && remotingCommand.getCode() != ResponseCode.CONTROLLER_NOT_LEADER) {
throw new RuntimeException("on broker heartbeat return invalid code, code: " + remotingCommand.getCode());
}
} catch (ExecutionException | InterruptedException | TimeoutException | RuntimeException e) {
log.error("on broker heartbeat through raft failed", e);
}
brokerChannelIdentityInfoMap.put(channel, brokerIdentityInfo);
}
@Override
public void onBrokerChannelClose(Channel channel) {
BrokerIdentityInfo brokerIdentityInfo = brokerChannelIdentityInfoMap.get(channel);
log.info("Channel {} inactive, broker identity info: {}", channel, brokerIdentityInfo);
if (brokerIdentityInfo != null) {
BrokerCloseChannelRequest requestHeader = new BrokerCloseChannelRequest(brokerIdentityInfo);
CompletableFuture<RemotingCommand> future = controller.onBrokerCloseChannel(requestHeader);
try {
RemotingCommand remotingCommand = future.get(5, java.util.concurrent.TimeUnit.SECONDS);
if (remotingCommand.getCode() != ResponseCode.SUCCESS) {
throw new RuntimeException("on broker close channel return invalid code, code: " + remotingCommand.getCode());
}
this.executor.submit(() -> notifyBrokerInActive(brokerIdentityInfo.getClusterName(), brokerIdentityInfo.getBrokerName(), brokerIdentityInfo.getBrokerId()));
brokerChannelIdentityInfoMap.remove(channel);
} catch (ExecutionException | InterruptedException | TimeoutException | RuntimeException e) {
log.error("on broker close channel through raft failed", e);
}
}
}
/**
* @param brokerIdentityInfo null means get broker live info of all brokers
*/
private Map<BrokerIdentityInfo, BrokerLiveInfo> getBrokerLiveInfo(BrokerIdentityInfo brokerIdentityInfo) {
GetBrokerLiveInfoRequest requestHeader;
if (brokerIdentityInfo == null) {
requestHeader = new GetBrokerLiveInfoRequest();
} else {
requestHeader = new GetBrokerLiveInfoRequest(brokerIdentityInfo);
}
CompletableFuture<RemotingCommand> future = controller.getBrokerLiveInfo(requestHeader);
try {
RemotingCommand remotingCommand = future.get(5, java.util.concurrent.TimeUnit.SECONDS);
if (remotingCommand.getCode() != ResponseCode.SUCCESS) {
throw new RuntimeException("get broker live info return invalid code, code: " + remotingCommand.getCode());
}
GetBrokerLiveInfoResponse getBrokerLiveInfoResponse = (GetBrokerLiveInfoResponse) remotingCommand.decodeCommandCustomHeader(GetBrokerLiveInfoResponse.class);
return JSON.parseObject(remotingCommand.getBody(), new TypeReference<Map<BrokerIdentityInfo, BrokerLiveInfo>>() {
}.getType());
} catch (Throwable e) {
log.error("get broker live info through raft failed", e);
}
return new HashMap<>();
}
private void scanNotActiveBroker() {
if (!controller.isLeaderState()) {
log.info("current node is not leader, skip scan not active broker");
return;
}
// if has not received any heartbeat from broker, we do not need to scan
if (this.firstReceivedHeartbeatTime + controllerConfig.getJraftConfig().getjRaftScanWaitTimeoutMs() < System.currentTimeMillis()) {
log.info("has not received any heartbeat from broker, skip scan not active broker");
return;
}
log.info("start scan not active broker");
CheckNotActiveBrokerRequest requestHeader = new CheckNotActiveBrokerRequest();
CompletableFuture<RemotingCommand> future = this.controller.checkNotActiveBroker(requestHeader);
try {
RemotingCommand remotingCommand = future.get(5, java.util.concurrent.TimeUnit.SECONDS);
if (remotingCommand.getCode() != ResponseCode.SUCCESS) {
throw new RuntimeException("check not active broker return invalid code, code: " + remotingCommand.getCode());
}
List<BrokerIdentityInfo> notActiveAndNeedReElectBrokerIdentityInfoList = JSON.parseObject(remotingCommand.getBody(), new TypeReference<List<BrokerIdentityInfo>>() {
}.getType());
if (notActiveAndNeedReElectBrokerIdentityInfoList != null && !notActiveAndNeedReElectBrokerIdentityInfoList.isEmpty()) {
notActiveAndNeedReElectBrokerIdentityInfoList.forEach(brokerIdentityInfo -> {
Iterator<Map.Entry<Channel, BrokerIdentityInfo>> iterator = brokerChannelIdentityInfoMap.entrySet().iterator();
Channel channel = null;
while (iterator.hasNext()) {
Map.Entry<Channel, BrokerIdentityInfo> entry = iterator.next();
if (entry.getValue().getBrokerId() == null) {
continue;
}
if (entry.getValue().equals(brokerIdentityInfo)) {
channel = entry.getKey();
RemotingHelper.closeChannel(entry.getKey());
iterator.remove();
break;
}
}
this.executor.submit(() -> notifyBrokerInActive(brokerIdentityInfo.getClusterName(), brokerIdentityInfo.getBrokerName(), brokerIdentityInfo.getBrokerId()));
log.warn("The broker channel {} expired, brokerInfo {}", channel, brokerIdentityInfo);
});
}
} catch (Throwable e) {
log.error("check not active broker through raft failed", e);
}
}
@Override
public BrokerLiveInfo getBrokerLiveInfo(String clusterName, String brokerName, Long brokerId) {
log.info("get broker live info, clusterName: {}, brokerName: {}, brokerId: {}", clusterName, brokerName, brokerId);
BrokerIdentityInfo brokerIdentityInfo = new BrokerIdentityInfo(clusterName, brokerName, brokerId);
Map<BrokerIdentityInfo, BrokerLiveInfo> brokerLiveInfoMap = getBrokerLiveInfo(brokerIdentityInfo);
return brokerLiveInfoMap.get(brokerIdentityInfo);
}
@Override
public boolean isBrokerActive(String clusterName, String brokerName, Long brokerId) {
BrokerLiveInfo info = null;
try {
info = getBrokerLiveInfo(clusterName, brokerName, brokerId);
} catch (RuntimeException e) {
log.error("get broker live info failed", e);
return false;
}
if (info != null) {
long last = info.getLastUpdateTimestamp();
long timeoutMillis = info.getHeartbeatTimeoutMillis();
return (last + timeoutMillis) >= System.currentTimeMillis();
}
return false;
}
@Override
public Map<String, Map<String, Integer>> getActiveBrokersNum() {
Map<String, Map<String, Integer>> map = new HashMap<>();
Map<BrokerIdentityInfo, BrokerLiveInfo> brokerLiveInfoMap = getBrokerLiveInfo(null);
brokerLiveInfoMap.keySet().stream()
.filter(brokerIdentity -> this.isBrokerActive(brokerIdentity.getClusterName(), brokerIdentity.getBrokerName(), brokerIdentity.getBrokerId()))
.forEach(id -> {
map.computeIfAbsent(id.getClusterName(), k -> new HashMap<>());
map.get(id.getClusterName()).compute(id.getBrokerName(), (broker, num) ->
num == null ? 0 : num + 1
);
});
return map;
}
private void notifyBrokerInActive(String clusterName, String brokerName, Long brokerId) {
log.info("Broker {}-{}-{} inactive", clusterName, brokerName, brokerId);
for (BrokerLifecycleListener listener : this.brokerLifecycleListeners) {
listener.onBrokerInactive(clusterName, brokerName, brokerId);
}
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/manager/BrokerReplicaInfo.java
+13 -6
View File
@@ -16,19 +16,21 @@
*/
package org.apache.rocketmq.controller.impl.manager;
import java.io.Serializable;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
/**
* Broker replicas info, mapping from brokerAddress to {brokerId, brokerHaAddress}.
*/
public class BrokerReplicaInfo {
public class BrokerReplicaInfo implements Serializable {
private final String clusterName;
private final String brokerName;
@@ -83,7 +85,9 @@ public class BrokerReplicaInfo {
}
public String getBrokerAddress(final Long brokerId) {
if (brokerId == null) return null;
if (brokerId == null) {
return null;
}
Pair<String, String> pair = this.brokerIdInfo.get(brokerId);
if (pair != null) {
return pair.getObject1();
@@ -92,7 +96,9 @@ public class BrokerReplicaInfo {
}
public String getBrokerRegisterCheckCode(final Long brokerId) {
if (brokerId == null) return null;
if (brokerId == null) {
return null;
}
Pair<String, String> pair = this.brokerIdInfo.get(brokerId);
if (pair != null) {
return pair.getObject2();
@@ -101,7 +107,8 @@ public class BrokerReplicaInfo {
}
public void updateBrokerAddress(final Long brokerId, final String brokerAddress) {
if (brokerId == null) return;
if (brokerId == null)
return;
Pair<String, String> oldPair = this.brokerIdInfo.get(brokerId);
if (oldPair != null) {
this.brokerIdInfo.put(brokerId, new Pair<>(brokerAddress, oldPair.getObject2()));
controller/src/main/java/org/apache/rocketmq/controller/impl/manager/RaftReplicasInfoManager.java
+236
View File
@@ -0,0 +1,236 @@
/*
* 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.controller.impl.manager;
import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.controller.helper.BrokerValidPredicate;
import org.apache.rocketmq.controller.impl.event.ControllerResult;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerIdentityInfo;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
import org.apache.rocketmq.controller.impl.task.BrokerCloseChannelRequest;
import org.apache.rocketmq.controller.impl.task.BrokerCloseChannelResponse;
import org.apache.rocketmq.controller.impl.task.CheckNotActiveBrokerRequest;
import org.apache.rocketmq.controller.impl.task.CheckNotActiveBrokerResponse;
import org.apache.rocketmq.controller.impl.task.GetBrokerLiveInfoRequest;
import org.apache.rocketmq.controller.impl.task.GetBrokerLiveInfoResponse;
import org.apache.rocketmq.controller.impl.task.RaftBrokerHeartBeatEventRequest;
import org.apache.rocketmq.controller.impl.task.RaftBrokerHeartBeatEventResponse;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import java.io.ByteArrayOutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
public class RaftReplicasInfoManager extends ReplicasInfoManager {
private static final Logger log = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private final Map<BrokerIdentityInfo/* brokerIdentity*/, BrokerLiveInfo> brokerLiveTable = new ConcurrentHashMap<>(256);
public RaftReplicasInfoManager(ControllerConfig controllerConfig) {
super(controllerConfig);
}
public ControllerResult<GetBrokerLiveInfoResponse> getBrokerLiveInfo(final GetBrokerLiveInfoRequest request) {
BrokerIdentityInfo brokerIdentityInfo = request.getBrokerIdentity();
ControllerResult<GetBrokerLiveInfoResponse> result = new ControllerResult<>(new GetBrokerLiveInfoResponse());
Map<BrokerIdentityInfo/* brokerIdentity*/, BrokerLiveInfo> resBrokerLiveTable = new HashMap<>();
if (brokerIdentityInfo == null || brokerIdentityInfo.isEmpty()) {
resBrokerLiveTable.putAll(this.brokerLiveTable);
} else {
if (brokerLiveTable.containsKey(brokerIdentityInfo)) {
resBrokerLiveTable.put(brokerIdentityInfo, brokerLiveTable.get(brokerIdentityInfo));
} else {
log.warn("GetBrokerLiveInfo failed, brokerIdentityInfo: {} not exist", brokerIdentityInfo);
result.setCodeAndRemark(ResponseCode.CONTROLLER_BROKER_LIVE_INFO_NOT_EXISTS, "brokerIdentityInfo not exist");
}
}
try {
result.setBody(JSON.toJSONBytes(resBrokerLiveTable));
} catch (Throwable e) {
log.error("json serialize resBrokerLiveTable {} error", resBrokerLiveTable, e);
result.setCodeAndRemark(ResponseCode.SYSTEM_ERROR, "serialize error");
}
return result;
}
public ControllerResult<RaftBrokerHeartBeatEventResponse> onBrokerHeartBeat(
RaftBrokerHeartBeatEventRequest request) {
BrokerIdentityInfo brokerIdentityInfo = request.getBrokerIdentityInfo();
BrokerLiveInfo brokerLiveInfo = request.getBrokerLiveInfo();
ControllerResult<RaftBrokerHeartBeatEventResponse> result = new ControllerResult<>(new RaftBrokerHeartBeatEventResponse());
BrokerLiveInfo prev = brokerLiveTable.computeIfAbsent(brokerIdentityInfo, identityInfo -> {
log.info("new broker registered, brokerIdentityInfo: {}", identityInfo);
return brokerLiveInfo;
});
prev.setLastUpdateTimestamp(brokerLiveInfo.getLastUpdateTimestamp());
prev.setHeartbeatTimeoutMillis(brokerLiveInfo.getHeartbeatTimeoutMillis());
prev.setElectionPriority(brokerLiveInfo.getElectionPriority());
if (brokerLiveInfo.getEpoch() > prev.getEpoch() || brokerLiveInfo.getEpoch() == prev.getEpoch() && brokerLiveInfo.getMaxOffset() > prev.getMaxOffset()) {
prev.setEpoch(brokerLiveInfo.getEpoch());
prev.setMaxOffset(brokerLiveInfo.getMaxOffset());
prev.setConfirmOffset(brokerLiveInfo.getConfirmOffset());
}
return result;
}
public ControllerResult<BrokerCloseChannelResponse> onBrokerCloseChannel(BrokerCloseChannelRequest request) {
BrokerIdentityInfo brokerIdentityInfo = request.getBrokerIdentityInfo();
ControllerResult<BrokerCloseChannelResponse> result = new ControllerResult<>(new BrokerCloseChannelResponse());
if (brokerIdentityInfo == null || brokerIdentityInfo.isEmpty()) {
log.warn("onBrokerCloseChannel failed, brokerIdentityInfo is null");
} else {
brokerLiveTable.remove(brokerIdentityInfo);
log.info("onBrokerCloseChannel success, brokerIdentityInfo: {}", brokerIdentityInfo);
}
return result;
}
public ControllerResult<CheckNotActiveBrokerResponse> checkNotActiveBroker(CheckNotActiveBrokerRequest request) {
List<BrokerIdentityInfo> notActiveBrokerIdentityInfoList = new ArrayList<>();
long checkTime = request.getCheckTimeMillis();
final Iterator<Map.Entry<BrokerIdentityInfo, BrokerLiveInfo>> iterator = this.brokerLiveTable.entrySet().iterator();
while (iterator.hasNext()) {
final Map.Entry<BrokerIdentityInfo, BrokerLiveInfo> next = iterator.next();
long last = next.getValue().getLastUpdateTimestamp();
long timeoutMillis = next.getValue().getHeartbeatTimeoutMillis();
if (checkTime - last > timeoutMillis) {
notActiveBrokerIdentityInfoList.add(next.getKey());
iterator.remove();
log.warn("Broker expired, brokerInfo {}, expired {}ms", next.getKey(), timeoutMillis);
}
}
List<String> needReElectBrokerNames = scanNeedReelectBrokerSets(new BrokerValidPredicate() {
@Override
public boolean check(String clusterName, String brokerName, Long brokerId) {
return !isBrokerActive(clusterName, brokerName, brokerId, checkTime);
}
});
Set<String> alreadyReportedBrokerName = notActiveBrokerIdentityInfoList.stream()
.map(BrokerIdentityInfo::getBrokerName)
.collect(Collectors.toSet());
// avoid to duplicate report, filter by name,
// because BrokerIdentityInfo in needReElectBrokerNames does not have brokerId or clusterName
notActiveBrokerIdentityInfoList.addAll(needReElectBrokerNames.stream()
.filter(brokerName -> !alreadyReportedBrokerName.contains(brokerName))
.map(brokerName -> new BrokerIdentityInfo(null, brokerName, null))
.collect(Collectors.toList()));
ControllerResult<CheckNotActiveBrokerResponse> result = new ControllerResult<>(new CheckNotActiveBrokerResponse());
try {
result.setBody(JSON.toJSONBytes(notActiveBrokerIdentityInfoList));
} catch (Throwable e) {
log.error("json serialize notActiveBrokerIdentityInfoList {} error", notActiveBrokerIdentityInfoList, e);
result.setCodeAndRemark(ResponseCode.SYSTEM_ERROR, "serialize error");
}
return result;
}
public boolean isBrokerActive(String clusterName, String brokerName, Long brokerId, long invokeTime) {
final BrokerLiveInfo info = this.brokerLiveTable.get(new BrokerIdentityInfo(clusterName, brokerName, brokerId));
if (info != null) {
long last = info.getLastUpdateTimestamp();
long timeoutMillis = info.getHeartbeatTimeoutMillis();
return (last + timeoutMillis) >= invokeTime;
}
return false;
}
public BrokerLiveInfo getBrokerLiveInfo(String clusterName, String brokerName, Long brokerId) {
return this.brokerLiveTable.get(new BrokerIdentityInfo(clusterName, brokerName, brokerId));
}
@Override
public byte[] serialize() throws Throwable {
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
final byte[] superSerialize = super.serialize();
putInt(outputStream, superSerialize.length);
outputStream.write(superSerialize);
putInt(outputStream, this.brokerLiveTable.size());
for (Map.Entry<BrokerIdentityInfo, BrokerLiveInfo> entry : brokerLiveTable.entrySet()) {
final byte[] brokerIdentityInfo = hessianSerialize(entry.getKey());
final byte[] brokerLiveInfo = hessianSerialize(entry.getValue());
putInt(outputStream, brokerIdentityInfo.length);
outputStream.write(brokerIdentityInfo);
putInt(outputStream, brokerLiveInfo.length);
outputStream.write(brokerLiveInfo);
}
return outputStream.toByteArray();
} catch (Throwable e) {
log.error("serialize replicaInfoTable or syncStateSetInfoTable error", e);
throw e;
}
}
@Override
public void deserializeFrom(byte[] data) throws Throwable {
int index = 0;
this.brokerLiveTable.clear();
try {
int superTableSize = getInt(data, index);
index += 4;
byte[] superTableData = new byte[superTableSize];
System.arraycopy(data, index, superTableData, 0, superTableSize);
super.deserializeFrom(superTableData);
index += superTableSize;
int brokerLiveTableSize = getInt(data, index);
index += 4;
for (int i = 0; i < brokerLiveTableSize; i++) {
int brokerIdentityInfoLength = getInt(data, index);
index += 4;
byte[] brokerIdentityInfoArray = new byte[brokerIdentityInfoLength];
System.arraycopy(data, index, brokerIdentityInfoArray, 0, brokerIdentityInfoLength);
BrokerIdentityInfo brokerIdentityInfo = (BrokerIdentityInfo) hessianDeserialize(brokerIdentityInfoArray);
index += brokerIdentityInfoLength;
int brokerLiveInfoLength = getInt(data, index);
index += 4;
byte[] brokerLiveInfoArray = new byte[brokerLiveInfoLength];
System.arraycopy(data, index, brokerLiveInfoArray, 0, brokerLiveInfoLength);
BrokerLiveInfo brokerLiveInfo = (BrokerLiveInfo) hessianDeserialize(brokerLiveInfoArray);
index += brokerLiveInfoLength;
this.brokerLiveTable.put(brokerIdentityInfo, brokerLiveInfo);
}
} catch (Throwable e) {
log.error("deserialize replicaInfoTable or syncStateSetInfoTable error", e);
throw e;
}
}
public static class BrokerValidPredicateWithInvokeTime implements BrokerValidPredicate {
private final long invokeTime;
private final RaftReplicasInfoManager raftBrokerHeartBeatManager;
public BrokerValidPredicateWithInvokeTime(long invokeTime, RaftReplicasInfoManager raftBrokerHeartBeatManager) {
this.invokeTime = invokeTime;
this.raftBrokerHeartBeatManager = raftBrokerHeartBeatManager;
}
@Override
public boolean check(String clusterName, String brokerName, Long brokerId) {
return raftBrokerHeartBeatManager.isBrokerActive(clusterName, brokerName, brokerId, invokeTime);
}
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManager.java
+136 -26
View File
@@ -16,17 +16,11 @@
*/
package org.apache.rocketmq.controller.impl.manager;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.caucho.hessian.io.Hessian2Input;
import com.caucho.hessian.io.Hessian2Output;
import com.caucho.hessian.io.SerializerFactory;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.MixAll;
@@ -50,11 +44,11 @@ import org.apache.rocketmq.remoting.protocol.body.ElectMasterResponseBody;
import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
@@ -62,6 +56,18 @@ import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextB
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerResponseHeader;
import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* The manager that manages the replicas info for all brokers. We can think of this class as the controller's memory
@@ -69,10 +75,32 @@ import org.apache.rocketmq.remoting.protocol.header.controller.register.Register
*/
public class ReplicasInfoManager {
private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
private final ControllerConfig controllerConfig;
protected static final SerializerFactory SERIALIZER_FACTORY = new SerializerFactory();
protected final ControllerConfig controllerConfig;
private final Map<String/* brokerName */, BrokerReplicaInfo> replicaInfoTable;
private final Map<String/* brokerName */, SyncStateInfo> syncStateSetInfoTable;
protected static byte[] hessianSerialize(Object object) throws IOException {
try (ByteArrayOutputStream bout = new ByteArrayOutputStream()) {
Hessian2Output hessianOut = new Hessian2Output(bout);
hessianOut.setSerializerFactory(SERIALIZER_FACTORY);
hessianOut.writeObject(object);
hessianOut.close();
return bout.toByteArray();
}
}
protected static Object hessianDeserialize(byte[] data) throws IOException {
try (ByteArrayInputStream bin = new ByteArrayInputStream(data, 0, data.length)) {
Hessian2Input hin = new Hessian2Input(bin);
hin.setSerializerFactory(new SerializerFactory());
Object o = hin.readObject();
hin.close();
return o;
}
}
public ReplicasInfoManager(final ControllerConfig config) {
this.controllerConfig = config;
this.replicaInfoTable = new ConcurrentHashMap<String, BrokerReplicaInfo>();
@@ -80,8 +108,8 @@ public class ReplicasInfoManager {
}
public ControllerResult<AlterSyncStateSetResponseHeader> alterSyncStateSet(
final AlterSyncStateSetRequestHeader request, final SyncStateSet syncStateSet,
final BrokerValidPredicate brokerAlivePredicate) {
final AlterSyncStateSetRequestHeader request, final SyncStateSet syncStateSet,
final BrokerValidPredicate brokerAlivePredicate) {
final String brokerName = request.getBrokerName();
final ControllerResult<AlterSyncStateSetResponseHeader> result = new ControllerResult<>(new AlterSyncStateSetResponseHeader());
final AlterSyncStateSetResponseHeader response = result.getResponse();
@@ -106,7 +134,7 @@ public class ReplicasInfoManager {
// Check master
if (syncStateInfo.getMasterBrokerId() == null || !syncStateInfo.getMasterBrokerId().equals(request.getMasterBrokerId())) {
String err = String.format("Rejecting alter syncStateSet request because the current leader is:{%s}, not {%s}",
syncStateInfo.getMasterBrokerId(), request.getMasterBrokerId());
syncStateInfo.getMasterBrokerId(), request.getMasterBrokerId());
LOGGER.error("{}", err);
result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_MASTER, err);
return result;
@@ -115,7 +143,7 @@ public class ReplicasInfoManager {
// Check master epoch
if (request.getMasterEpoch() != syncStateInfo.getMasterEpoch()) {
String err = String.format("Rejecting alter syncStateSet request because the current master epoch is:{%d}, not {%d}",
syncStateInfo.getMasterEpoch(), request.getMasterEpoch());
syncStateInfo.getMasterEpoch(), request.getMasterEpoch());
LOGGER.error("{}", err);
result.setCodeAndRemark(ResponseCode.CONTROLLER_FENCED_MASTER_EPOCH, err);
return result;
@@ -124,7 +152,7 @@ public class ReplicasInfoManager {
// Check syncStateSet epoch
if (syncStateSet.getSyncStateSetEpoch() != syncStateInfo.getSyncStateSetEpoch()) {
String err = String.format("Rejecting alter syncStateSet request because the current syncStateSet epoch is:{%d}, not {%d}",
syncStateInfo.getSyncStateSetEpoch(), syncStateSet.getSyncStateSetEpoch());
syncStateInfo.getSyncStateSetEpoch(), syncStateSet.getSyncStateSetEpoch());
LOGGER.error("{}", err);
result.setCodeAndRemark(ResponseCode.CONTROLLER_FENCED_SYNC_STATE_SET_EPOCH, err);
return result;
@@ -163,7 +191,7 @@ public class ReplicasInfoManager {
}
public ControllerResult<ElectMasterResponseHeader> electMaster(final ElectMasterRequestHeader request,
final ElectPolicy electPolicy) {
final ElectPolicy electPolicy) {
final String brokerName = request.getBrokerName();
final Long brokerId = request.getBrokerId();
final ControllerResult<ElectMasterResponseHeader> result = new ControllerResult<>(new ElectMasterResponseHeader());
@@ -188,7 +216,7 @@ public class ReplicasInfoManager {
}
// elect by policy
if (newMaster == null) {
if (newMaster == null || newMaster == -1) {
// we should assign this assignedBrokerId when the brokerAddress need to be elected by force
Long assignedBrokerId = request.getDesignateElect() ? brokerId : null;
newMaster = electPolicy.elect(brokerReplicaInfo.getClusterName(), brokerReplicaInfo.getBrokerName(), syncStateSet, allReplicaBrokers, oldMaster, assignedBrokerId);
@@ -230,18 +258,20 @@ public class ReplicasInfoManager {
result.setBody(responseBody.encode());
final ElectMasterEvent event = new ElectMasterEvent(brokerName, newMaster);
result.addEvent(event);
LOGGER.info("Elect new master {} for broker {}", newMaster, brokerName);
return result;
}
// If elect failed and the electMaster is triggered by controller (we can figure it out by brokerAddress),
// we still need to apply an ElectMasterEvent to tell the statemachine
// that the master was shutdown and no new master was elected.
if (request.getBrokerId() == null) {
if (request.getBrokerId() == null || request.getBrokerId() == -1) {
final ElectMasterEvent event = new ElectMasterEvent(false, brokerName);
result.addEvent(event);
result.setCodeAndRemark(ResponseCode.CONTROLLER_MASTER_NOT_AVAILABLE, "Old master has down and failed to elect a new broker master");
} else {
result.setCodeAndRemark(ResponseCode.CONTROLLER_ELECT_MASTER_FAILED, "Failed to elect a new master");
}
LOGGER.warn("Failed to elect a new master for broker {}", brokerName);
return result;
}
@@ -301,7 +331,8 @@ public class ReplicasInfoManager {
return result;
}
public ControllerResult<RegisterBrokerToControllerResponseHeader> registerBroker(final RegisterBrokerToControllerRequestHeader request, final BrokerValidPredicate alivePredicate) {
public ControllerResult<RegisterBrokerToControllerResponseHeader> registerBroker(
final RegisterBrokerToControllerRequestHeader request, final BrokerValidPredicate alivePredicate) {
final String brokerAddress = request.getBrokerAddress();
final String brokerName = request.getBrokerName();
final String clusterName = request.getClusterName();
@@ -353,7 +384,8 @@ public class ReplicasInfoManager {
return result;
}
public ControllerResult<Void> getSyncStateData(final List<String> brokerNames, final BrokerValidPredicate brokerAlivePredicate) {
public ControllerResult<Void> getSyncStateData(final List<String> brokerNames,
final BrokerValidPredicate brokerAlivePredicate) {
final ControllerResult<Void> result = new ControllerResult<>();
final BrokerReplicasInfo brokerReplicasInfo = new BrokerReplicasInfo();
for (String brokerName : brokerNames) {
@@ -382,7 +414,7 @@ public class ReplicasInfoManager {
});
final BrokerReplicasInfo.ReplicasInfo inSyncState = new BrokerReplicasInfo.ReplicasInfo(masterBrokerId, brokerReplicaInfo.getBrokerAddress(masterBrokerId), syncStateInfo.getMasterEpoch(), syncStateInfo.getSyncStateSetEpoch(),
inSyncReplicas, notInSyncReplicas);
inSyncReplicas, notInSyncReplicas);
brokerReplicasInfo.addReplicaInfo(brokerName, inSyncState);
}
}
@@ -391,7 +423,7 @@ public class ReplicasInfoManager {
}
public ControllerResult<Void> cleanBrokerData(final CleanControllerBrokerDataRequestHeader requestHeader,
final BrokerValidPredicate validPredicate) {
final BrokerValidPredicate validPredicate) {
final ControllerResult<Void> result = new ControllerResult<>();
final String clusterName = requestHeader.getClusterName();
@@ -451,7 +483,6 @@ public class ReplicasInfoManager {
return needReelectBrokerSets;
}
/**
* Apply events to memory statemachine.
*
@@ -576,4 +607,83 @@ public class ReplicasInfoManager {
return this.replicaInfoTable.containsKey(brokerName) && this.syncStateSetInfoTable.containsKey(brokerName);
}
protected void putInt(ByteArrayOutputStream outputStream, int value) {
outputStream.write((byte) (value >>> 24));
outputStream.write((byte) (value >>> 16));
outputStream.write((byte) (value >>> 8));
outputStream.write((byte) value);
}
protected int getInt(byte[] memory, int index) {
return memory[index] << 24 | (memory[index + 1] & 0xFF) << 16 | (memory[index + 2] & 0xFF) << 8 | memory[index + 3] & 0xFF;
}
public byte[] serialize() throws Throwable {
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
putInt(outputStream, this.replicaInfoTable.size());
for (Map.Entry<String, BrokerReplicaInfo> entry : replicaInfoTable.entrySet()) {
final byte[] brokerName = entry.getKey().getBytes(StandardCharsets.UTF_8);
byte[] brokerReplicaInfo = hessianSerialize(entry.getValue());
putInt(outputStream, brokerName.length);
outputStream.write(brokerName);
putInt(outputStream, brokerReplicaInfo.length);
outputStream.write(brokerReplicaInfo);
}
putInt(outputStream, this.syncStateSetInfoTable.size());
for (Map.Entry<String, SyncStateInfo> entry : syncStateSetInfoTable.entrySet()) {
final byte[] brokerName = entry.getKey().getBytes(StandardCharsets.UTF_8);
byte[] syncStateInfo = hessianSerialize(entry.getValue());
putInt(outputStream, brokerName.length);
outputStream.write(brokerName);
putInt(outputStream, syncStateInfo.length);
outputStream.write(syncStateInfo);
}
return outputStream.toByteArray();
} catch (Throwable e) {
LOGGER.error("serialize replicaInfoTable or syncStateSetInfoTable error", e);
throw e;
}
}
public void deserializeFrom(byte[] data) throws Throwable {
int index = 0;
this.replicaInfoTable.clear();
this.syncStateSetInfoTable.clear();
try {
int replicaInfoTableSize = getInt(data, index);
index += 4;
for (int i = 0; i < replicaInfoTableSize; i++) {
int brokerNameLength = getInt(data, index);
index += 4;
String brokerName = new String(data, index, brokerNameLength, StandardCharsets.UTF_8);
index += brokerNameLength;
int brokerReplicaInfoLength = getInt(data, index);
index += 4;
byte[] brokerReplicaInfoArray = new byte[brokerReplicaInfoLength];
System.arraycopy(data, index, brokerReplicaInfoArray, 0, brokerReplicaInfoLength);
BrokerReplicaInfo brokerReplicaInfo = (BrokerReplicaInfo) hessianDeserialize(brokerReplicaInfoArray);
index += brokerReplicaInfoLength;
this.replicaInfoTable.put(brokerName, brokerReplicaInfo);
}
int syncStateSetInfoTableSize = getInt(data, index);
index += 4;
for (int i = 0; i < syncStateSetInfoTableSize; i++) {
int brokerNameLength = getInt(data, index);
index += 4;
String brokerName = new String(data, index, brokerNameLength, StandardCharsets.UTF_8);
index += brokerNameLength;
int syncStateInfoLength = getInt(data, index);
index += 4;
byte[] syncStateInfoArray = new byte[syncStateInfoLength];
System.arraycopy(data, index, syncStateInfoArray, 0, syncStateInfoLength);
SyncStateInfo syncStateInfo = (SyncStateInfo) hessianDeserialize(syncStateInfoArray);
index += syncStateInfoLength;
this.syncStateSetInfoTable.put(brokerName, syncStateInfo);
}
} catch (Throwable e) {
LOGGER.error("deserialize replicaInfoTable or syncStateSetInfoTable error", e);
throw e;
}
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/manager/SyncStateInfo.java
+2 -1
View File
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.controller.impl.manager;
import java.io.Serializable;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
@@ -24,7 +25,7 @@ import java.util.concurrent.atomic.AtomicInteger;
/**
* Manages the syncStateSet of broker replicas.
*/
public class SyncStateInfo {
public class SyncStateInfo implements Serializable {
private final String clusterName;
private final String brokerName;
private final AtomicInteger masterEpoch;
controller/src/main/java/org/apache/rocketmq/controller/impl/task/BrokerCloseChannelRequest.java
+69
View File
@@ -0,0 +1,69 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerIdentityInfo;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNullable;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class BrokerCloseChannelRequest implements CommandCustomHeader {
@CFNullable
private String clusterName;
@CFNullable
private String brokerName;
@CFNullable
private Long brokerId;
public BrokerCloseChannelRequest() {
this.clusterName = null;
this.brokerName = null;
this.brokerId = null;
}
public BrokerCloseChannelRequest(BrokerIdentityInfo brokerIdentityInfo) {
this.clusterName = brokerIdentityInfo.getClusterName();
this.brokerName = brokerIdentityInfo.getBrokerName();
this.brokerId = brokerIdentityInfo.getBrokerId();
}
public BrokerIdentityInfo getBrokerIdentityInfo() {
return new BrokerIdentityInfo(this.clusterName, this.brokerName, this.brokerId);
}
public void setBrokerIdentityInfo(BrokerIdentityInfo brokerIdentityInfo) {
this.clusterName = brokerIdentityInfo.getClusterName();
this.brokerName = brokerIdentityInfo.getBrokerName();
this.brokerId = brokerIdentityInfo.getBrokerId();
}
@Override
public void checkFields() throws RemotingCommandException {
}
@Override
public String toString() {
return "BrokerCloseChannelRequest{" +
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerId=" + brokerId +
'}';
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/task/BrokerCloseChannelResponse.java
+35
View File
@@ -0,0 +1,35 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class BrokerCloseChannelResponse implements CommandCustomHeader {
public BrokerCloseChannelResponse() {
}
@Override
public void checkFields() throws RemotingCommandException {
}
@Override
public String toString() {
return "BrokerCloseChannelResponse{}";
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/task/CheckNotActiveBrokerRequest.java
+43
View File
@@ -0,0 +1,43 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class CheckNotActiveBrokerRequest implements CommandCustomHeader {
private final Long checkTimeMillis = System.currentTimeMillis();
public CheckNotActiveBrokerRequest() {
}
public Long getCheckTimeMillis() {
return checkTimeMillis;
}
@Override
public void checkFields() throws RemotingCommandException {
}
@Override
public String toString() {
return "CheckNotActiveBrokerRequest{" +
"checkTimeMillis=" + checkTimeMillis +
'}';
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/task/CheckNotActiveBrokerResponse.java
+35
View File
@@ -0,0 +1,35 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class CheckNotActiveBrokerResponse implements CommandCustomHeader {
public CheckNotActiveBrokerResponse() {
}
@Override
public void checkFields() throws RemotingCommandException {
}
@Override
public String toString() {
return "CheckNotActiveBrokerResponse{}";
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/task/GetBrokerLiveInfoRequest.java
+66
View File
@@ -0,0 +1,66 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerIdentityInfo;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class GetBrokerLiveInfoRequest implements CommandCustomHeader {
private String clusterName;
private String brokerName;
private Long brokerId;
public GetBrokerLiveInfoRequest() {
this.clusterName = null;
this.brokerName = null;
this.brokerId = null;
}
/**
* @param brokerIdentity The BrokerIdentityInfo that needs to be queried, if it is null, it means obtaining BrokerLiveInfo for all brokers
*/
public GetBrokerLiveInfoRequest(BrokerIdentityInfo brokerIdentity) {
this.clusterName = brokerIdentity.getClusterName();
this.brokerName = brokerIdentity.getBrokerName();
this.brokerId = brokerIdentity.getBrokerId();
}
public BrokerIdentityInfo getBrokerIdentity() {
return new BrokerIdentityInfo(this.clusterName, this.brokerName, this.brokerId);
}
public void setBrokerIdentity(BrokerIdentityInfo brokerIdentity) {
this.clusterName = brokerIdentity.getClusterName();
this.brokerName = brokerIdentity.getBrokerName();
this.brokerId = brokerIdentity.getBrokerId();
}
@Override
public void checkFields() throws RemotingCommandException {
}
@Override
public String toString() {
return "GetBrokerLiveInfoRequest{" +
"brokerIdentity=" + getBrokerIdentity() +
'}';
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/task/GetBrokerLiveInfoResponse.java
+35
View File
@@ -0,0 +1,35 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class GetBrokerLiveInfoResponse implements CommandCustomHeader {
public GetBrokerLiveInfoResponse() {
}
@Override
public void checkFields() throws RemotingCommandException {
}
@Override
public String toString() {
return "GetBrokerLiveInfoResponse{}";
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/task/GetSyncStateDataRequest.java
+44
View File
@@ -0,0 +1,44 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class GetSyncStateDataRequest implements CommandCustomHeader {
private final Long invokeTime = System.currentTimeMillis();
@Override
public void checkFields() throws RemotingCommandException {
}
public GetSyncStateDataRequest() {
}
public Long getInvokeTime() {
return invokeTime;
}
@Override
public String toString() {
return "GetSyncStateDataRequest{" +
"invokeTime=" + invokeTime +
'}';
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/task/RaftBrokerHeartBeatEventRequest.java
+100
View File
@@ -0,0 +1,100 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerIdentityInfo;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class RaftBrokerHeartBeatEventRequest implements CommandCustomHeader {
// brokerIdentityInfo
private String clusterNameIdentityInfo;
private String brokerNameIdentityInfo;
private Long brokerIdIdentityInfo;
// brokerLiveInfo
private String brokerName;
private String brokerAddr;
private Long heartbeatTimeoutMillis;
private Long brokerId;
private Long lastUpdateTimestamp;
private Integer epoch;
private Long maxOffset;
private Long confirmOffset;
private Integer electionPriority;
public RaftBrokerHeartBeatEventRequest() {
}
public RaftBrokerHeartBeatEventRequest(BrokerIdentityInfo brokerIdentityInfo, BrokerLiveInfo brokerLiveInfo) {
this.clusterNameIdentityInfo = brokerIdentityInfo.getClusterName();
this.brokerNameIdentityInfo = brokerIdentityInfo.getBrokerName();
this.brokerIdIdentityInfo = brokerIdentityInfo.getBrokerId();
this.brokerName = brokerLiveInfo.getBrokerName();
this.brokerAddr = brokerLiveInfo.getBrokerAddr();
this.heartbeatTimeoutMillis = brokerLiveInfo.getHeartbeatTimeoutMillis();
this.brokerId = brokerLiveInfo.getBrokerId();
this.lastUpdateTimestamp = brokerLiveInfo.getLastUpdateTimestamp();
this.epoch = brokerLiveInfo.getEpoch();
this.maxOffset = brokerLiveInfo.getMaxOffset();
this.confirmOffset = brokerLiveInfo.getConfirmOffset();
this.electionPriority = brokerLiveInfo.getElectionPriority();
}
public BrokerIdentityInfo getBrokerIdentityInfo() {
return new BrokerIdentityInfo(clusterNameIdentityInfo, brokerNameIdentityInfo, brokerIdIdentityInfo);
}
public void setBrokerIdentityInfo(BrokerIdentityInfo brokerIdentityInfo) {
this.clusterNameIdentityInfo = brokerIdentityInfo.getClusterName();
this.brokerNameIdentityInfo = brokerIdentityInfo.getBrokerName();
this.brokerIdIdentityInfo = brokerIdentityInfo.getBrokerId();
}
public BrokerLiveInfo getBrokerLiveInfo() {
return new BrokerLiveInfo(brokerName, brokerAddr, brokerId, lastUpdateTimestamp, heartbeatTimeoutMillis, null, epoch, maxOffset, electionPriority, confirmOffset);
}
public void setBrokerLiveInfo(BrokerLiveInfo brokerLiveInfo) {
this.brokerName = brokerLiveInfo.getBrokerName();
this.brokerAddr = brokerLiveInfo.getBrokerAddr();
this.heartbeatTimeoutMillis = brokerLiveInfo.getHeartbeatTimeoutMillis();
this.brokerId = brokerLiveInfo.getBrokerId();
this.lastUpdateTimestamp = brokerLiveInfo.getLastUpdateTimestamp();
this.epoch = brokerLiveInfo.getEpoch();
this.maxOffset = brokerLiveInfo.getMaxOffset();
this.confirmOffset = brokerLiveInfo.getConfirmOffset();
this.electionPriority = brokerLiveInfo.getElectionPriority();
}
@Override
public void checkFields() throws RemotingCommandException {
}
@Override
public String toString() {
return "RaftBrokerHeartBeatEventRequest{" +
"brokerIdentityInfo=" + getBrokerIdentityInfo() +
", brokerLiveInfo=" + getBrokerLiveInfo() +
"}";
}
}
controller/src/main/java/org/apache/rocketmq/controller/impl/task/RaftBrokerHeartBeatEventResponse.java
+35
View File
@@ -0,0 +1,35 @@
/*
* 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.controller.impl.task;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
public class RaftBrokerHeartBeatEventResponse implements CommandCustomHeader {
public RaftBrokerHeartBeatEventResponse() {
}
@Override
public void checkFields() throws RemotingCommandException {
}
@Override
public String toString() {
return "RaftBrokerHeartBeatEventResponse{}";
}
}
controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsConstant.java
+1 -1
View File
@@ -62,7 +62,6 @@ public class ControllerMetricsConstant {
public static final String LABEL_ELECTION_RESULT = "election_result";
public enum RequestType {
CONTROLLER_ALTER_SYNC_STATE_SET(RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET),
@@ -112,6 +111,7 @@ public class ControllerMetricsConstant {
SUCCESS,
FAILED,
TIMEOUT;
public String getLowerCaseName() {
return this.name().toLowerCase();
}
controller/src/main/java/org/apache/rocketmq/controller/metrics/ControllerMetricsManager.java
+16 -9
View File
@@ -41,12 +41,6 @@ import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
import io.opentelemetry.sdk.metrics.export.MetricExporter;
import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
import io.opentelemetry.sdk.resources.Resource;
import java.io.File;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.UtilAll;
@@ -61,6 +55,13 @@ import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.slf4j.bridge.SLF4JBridgeHandler;
import java.io.File;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.AGGREGATION_DELTA;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.COUNTER_DLEDGER_OP_TOTAL;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.COUNTER_ELECTION_TOTAL;
@@ -165,9 +166,15 @@ public class ControllerMetricsManager {
private ControllerMetricsManager(ControllerManager controllerManager) {
this.controllerManager = controllerManager;
this.config = this.controllerManager.getControllerConfig();
this.LABEL_MAP.put(LABEL_ADDRESS, this.config.getDLedgerAddress());
this.LABEL_MAP.put(LABEL_GROUP, this.config.getControllerDLegerGroup());
this.LABEL_MAP.put(LABEL_PEER_ID, this.config.getControllerDLegerSelfId());
if (config.getControllerType().equals(ControllerConfig.JRAFT_CONTROLLER)) {
this.LABEL_MAP.put(LABEL_ADDRESS, this.config.getJraftConfig().getjRaftAddress());
this.LABEL_MAP.put(LABEL_GROUP, this.config.getJraftConfig().getjRaftGroupId());
this.LABEL_MAP.put(LABEL_PEER_ID, this.config.getJraftConfig().getjRaftServerId());
} else {
this.LABEL_MAP.put(LABEL_ADDRESS, this.config.getDLedgerAddress());
this.LABEL_MAP.put(LABEL_GROUP, this.config.getControllerDLegerGroup());
this.LABEL_MAP.put(LABEL_PEER_ID, this.config.getControllerDLegerSelfId());
}
this.init();
}
controller/src/main/java/org/apache/rocketmq/controller/processor/ControllerRequestProcessor.java
+4 -4
View File
@@ -44,14 +44,14 @@ import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.RoleChangeNotifyEntry;
import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.BrokerHeartbeatRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_REQUEST_HANDLE_STATUS;
import static org.apache.rocketmq.controller.metrics.ControllerMetricsConstant.LABEL_REQUEST_TYPE;
controller/src/main/resources/rmq.controller.logback.xml
+48 -8
View File
@@ -19,10 +19,13 @@
<configuration scan="true" scanPeriod="30 seconds">
<appender name="DefaultAppender"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}controller_default.log</file>
<file>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}controller_default.log
</file>
<append>true</append>
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}controller_default.%i.log.gz</fileNamePattern>
<fileNamePattern>
${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}controller_default.%i.log.gz
</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>5</maxIndex>
</rollingPolicy>
@@ -41,7 +44,9 @@
<file>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}dledger.log</file>
<append>true</append>
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}dledger.%i.log.gz</fileNamePattern>
<fileNamePattern>
${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}dledger.%i.log.gz
</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>5</maxIndex>
</rollingPolicy>
@@ -60,12 +65,41 @@
<discardingThreshold>0</discardingThreshold>
</appender>
<appender name="RocketmqTrafficAppender_inner"
<appender name="JRaftAppender_inner"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}controller_traffic.log</file>
<file>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}jraft.log</file>
<append>true</append>
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}controller_traffic.%i.log.gz</fileNamePattern>
<fileNamePattern>
${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}jraft.%i.log.gz
</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>5</maxIndex>
</rollingPolicy>
<triggeringPolicy
class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
<maxFileSize>100MB</maxFileSize>
</triggeringPolicy>
<encoder>
<pattern>%d{yyy-MM-dd HH:mm:ss,GMT+8} %p %t - %m%n</pattern>
<charset class="java.nio.charset.Charset">UTF-8</charset>
</encoder>
</appender>
<appender name="JRaftAppender" class="ch.qos.logback.classic.AsyncAppender">
<appender-ref ref="JRaftAppender_inner"/>
<discardingThreshold>0</discardingThreshold>
</appender>
<appender name="RocketmqTrafficAppender_inner"
class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}controller_traffic.log
</file>
<append>true</append>
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>
${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}controller_traffic.%i.log.gz
</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>10</maxIndex>
</rollingPolicy>
@@ -87,7 +121,9 @@
<file>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}controller.log</file>
<append>true</append>
<rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
<fileNamePattern>${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}controller.%i.log.gz</fileNamePattern>
<fileNamePattern>
${user.home}${file.separator}logs${file.separator}rocketmqlogs${file.separator}otherdays${file.separator}controller.%i.log.gz
</fileNamePattern>
<minIndex>1</minIndex>
<maxIndex>5</maxIndex>
</rollingPolicy>
@@ -132,12 +168,16 @@
<appender-ref ref="DLedgerAppender"/>
</logger>
<logger name="com.alipay.sofa.jraft" additivity="false" level="INFO">
<appender-ref ref="JRaftAppender"/>
</logger>
<logger name="RocketmqControllerConsole" additivity="false" level="INFO">
<appender-ref ref="STDOUT"/>
</logger>
<logger name="RocketmqTraffic" additivity="false" level="INFO">
<appender-ref ref="RocketmqTrafficAppender" />
<appender-ref ref="RocketmqTrafficAppender"/>
</logger>
<root level="INFO">
controller/src/test/java/org/apache/rocketmq/controller/ControllerManagerTest.java
+17 -15
View File
@@ -16,14 +16,6 @@
*/
package org.apache.rocketmq.controller;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.controller.impl.DLedgerController;
@@ -35,6 +27,9 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
@@ -42,14 +37,19 @@ import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextB
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.BrokerHeartbeatRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@@ -131,7 +131,8 @@ public class ControllerManagerTest {
*/
public void registerBroker(
final String controllerAddress, final String clusterName,
final String brokerName, final Long brokerId, final String brokerAddress, final Long expectMasterBrokerId, final RemotingClient client) throws Exception {
final String brokerName, final Long brokerId, final String brokerAddress, final Long expectMasterBrokerId,
final RemotingClient client) throws Exception {
// Get next brokerId;
final GetNextBrokerIdRequestHeader getNextBrokerIdRequestHeader = new GetNextBrokerIdRequestHeader(clusterName, brokerName);
final RemotingCommand getNextBrokerIdRequest = RemotingCommand.createRequestCommand(RequestCode.CONTROLLER_GET_NEXT_BROKER_ID, getNextBrokerIdRequestHeader);
@@ -166,8 +167,9 @@ public class ControllerManagerTest {
return response;
}
public void sendHeartbeat(final String controllerAddress, final String clusterName, final String brokerName, final Long brokerId,
final String brokerAddress, final Long timeout, final RemotingClient client) throws Exception {
public void sendHeartbeat(final String controllerAddress, final String clusterName, final String brokerName,
final Long brokerId,
final String brokerAddress, final Long timeout, final RemotingClient client) throws Exception {
final BrokerHeartbeatRequestHeader heartbeatRequestHeader0 = new BrokerHeartbeatRequestHeader();
heartbeatRequestHeader0.setBrokerId(brokerId);
heartbeatRequestHeader0.setClusterName(clusterName);
controller/src/test/java/org/apache/rocketmq/controller/ControllerRequestProcessorTest.java
+3 -2
View File
@@ -17,8 +17,6 @@
package org.apache.rocketmq.controller;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.controller.processor.ControllerRequestProcessor;
@@ -30,6 +28,9 @@ import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.junit.Before;
import org.junit.Test;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
public class ControllerRequestProcessorTest {
controller/src/test/java/org/apache/rocketmq/controller/impl/DLedgerControllerTest.java
+12 -12
View File
@@ -16,18 +16,6 @@
*/
package org.apache.rocketmq.controller.impl;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.controller.Controller;
@@ -49,6 +37,18 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_BROKER_NAME;
import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_CLUSTER_NAME;
import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_IP;
controller/src/test/java/org/apache/rocketmq/controller/impl/DefaultBrokerHeartbeatManagerTest.java
+5 -4
View File
@@ -16,14 +16,15 @@
*/
package org.apache.rocketmq.controller.impl;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.controller.BrokerHeartbeatManager;
import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
import org.junit.Before;
import org.junit.Test;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertTrue;
public class DefaultBrokerHeartbeatManagerTest {
@@ -44,8 +45,8 @@ public class DefaultBrokerHeartbeatManagerTest {
this.heartbeatManager.registerBrokerLifecycleListener((clusterName, brokerName, brokerId) -> {
latch.countDown();
});
this.heartbeatManager.onBrokerHeartbeat("cluster1", "broker1", "127.0.0.1:7000", 1L,3000L, null,
1, 1L,-1L, 0);
this.heartbeatManager.onBrokerHeartbeat("cluster1", "broker1", "127.0.0.1:7000", 1L, 3000L, null,
1, 1L, -1L, 0);
assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
this.heartbeatManager.shutdown();
}
controller/src/test/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManagerTest.java
+65 -20
View File
@@ -16,30 +16,25 @@
*/
package org.apache.rocketmq.controller.impl.manager;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import org.apache.rocketmq.common.ControllerConfig;
import org.apache.rocketmq.controller.elect.ElectPolicy;
import org.apache.rocketmq.controller.elect.impl.DefaultElectPolicy;
import org.apache.rocketmq.controller.helper.BrokerValidPredicate;
import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
import org.apache.rocketmq.controller.impl.event.ControllerResult;
import org.apache.rocketmq.controller.impl.event.ElectMasterEvent;
import org.apache.rocketmq.controller.impl.event.EventMessage;
import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
@@ -50,6 +45,14 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_BROKER_NAME;
import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_CLUSTER_NAME;
import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_IP;
@@ -67,7 +70,6 @@ public class ReplicasInfoManagerTest {
private ControllerConfig config;
@Before
public void init() {
this.config = new ControllerConfig();
@@ -93,7 +95,7 @@ public class ReplicasInfoManagerTest {
}
public void registerNewBroker(String clusterName, String brokerName, String brokerAddress,
Long exceptBrokerId, Long exceptMasterBrokerId) {
Long exceptBrokerId, Long exceptMasterBrokerId) {
// Get next brokerId
final GetNextBrokerIdRequestHeader getNextBrokerIdRequestHeader = new GetNextBrokerIdRequestHeader(clusterName, brokerName);
@@ -131,11 +133,14 @@ public class ReplicasInfoManagerTest {
assertEquals(exceptMasterBrokerId, registerSuccessResult.getResponse().getMasterBrokerId());
}
public void brokerElectMaster(String clusterName, Long brokerId, String brokerName, String brokerAddress, boolean isFirstTryElect, boolean expectToBeElected) {
this.brokerElectMaster(clusterName, brokerId, brokerName, brokerAddress, isFirstTryElect,expectToBeElected, (a, b, c) -> true);
public void brokerElectMaster(String clusterName, Long brokerId, String brokerName, String brokerAddress,
boolean isFirstTryElect, boolean expectToBeElected) {
this.brokerElectMaster(clusterName, brokerId, brokerName, brokerAddress, isFirstTryElect, expectToBeElected, (a, b, c) -> true);
}
public void brokerElectMaster(String clusterName, Long brokerId, String brokerName, String brokerAddress, boolean isFirstTryElect, boolean expectToBeElected, BrokerValidPredicate validPredicate) {
public void brokerElectMaster(String clusterName, Long brokerId, String brokerName, String brokerAddress,
boolean isFirstTryElect, boolean expectToBeElected, BrokerValidPredicate validPredicate) {
final GetReplicaInfoResponseHeader replicaInfoBefore = this.replicasInfoManager.getReplicaInfo(new GetReplicaInfoRequestHeader(brokerName)).getResponse();
BrokerReplicasInfo.ReplicasInfo syncStateSetInfo = getReplicasInfo(brokerName);
@@ -174,7 +179,7 @@ public class ReplicasInfoManagerTest {
// a new master can be elected successfully
assertEquals(ResponseCode.SUCCESS, result.getResponseCode());
assertEquals(replicaInfoBefore.getMasterEpoch() + 1, replicaInfoAfter.getMasterEpoch().intValue());
if (expectToBeElected) {
assertEquals(brokerAddress, response.getMasterAddress());
assertEquals(brokerId, response.getMasterBrokerId());
@@ -194,7 +199,7 @@ public class ReplicasInfoManagerTest {
final AlterSyncStateSetRequestHeader alterRequest =
new AlterSyncStateSetRequestHeader(brokerName, brokerId, masterEpoch);
final ControllerResult<AlterSyncStateSetResponseHeader> result = this.replicasInfoManager.alterSyncStateSet(alterRequest,
new SyncStateSet(newSyncStateSet, syncStateSetEpoch), (cluster, brokerName1, brokerId1) -> true);
new SyncStateSet(newSyncStateSet, syncStateSetEpoch), (cluster, brokerName1, brokerId1) -> true);
apply(result.getEvents());
final ControllerResult<GetReplicaInfoResponseHeader> resp = this.replicasInfoManager.getReplicaInfo(new GetReplicaInfoRequestHeader(brokerName));
@@ -367,7 +372,7 @@ public class ReplicasInfoManagerTest {
mockMetaData();
final ElectMasterRequestHeader request = ElectMasterRequestHeader.ofControllerTrigger(DEFAULT_BROKER_NAME);
final ControllerResult<ElectMasterResponseHeader> cResult = this.replicasInfoManager.electMaster(request,
new DefaultElectPolicy((cluster, brokerName, brokerId) -> !brokerId.equals(1L), null));
new DefaultElectPolicy((cluster, brokerName, brokerId) -> !brokerId.equals(1L), null));
final ElectMasterResponseHeader response = cResult.getResponse();
assertEquals(2, response.getMasterEpoch().intValue());
assertNotEquals(1L, response.getMasterBrokerId().longValue());
@@ -383,20 +388,20 @@ public class ReplicasInfoManagerTest {
// test admin try to elect a assignedMaster, but it isn't alive
final ElectMasterRequestHeader assignRequest = ElectMasterRequestHeader.ofAdminTrigger(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, 1L);
final ControllerResult<ElectMasterResponseHeader> cResult1 = this.replicasInfoManager.electMaster(assignRequest,
new DefaultElectPolicy((cluster, brokerName, brokerId) -> !brokerId.equals(1L), null));
new DefaultElectPolicy((cluster, brokerName, brokerId) -> !brokerId.equals(1L), null));
assertEquals(cResult1.getResponseCode(), ResponseCode.CONTROLLER_ELECT_MASTER_FAILED);
// test admin try to elect a assignedMaster but old master still alive, and the old master is equals to assignedMaster
final ElectMasterRequestHeader assignRequest1 = ElectMasterRequestHeader.ofAdminTrigger(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, response.getMasterBrokerId());
final ControllerResult<ElectMasterResponseHeader> cResult2 = this.replicasInfoManager.electMaster(assignRequest1,
new DefaultElectPolicy((cluster, brokerName, brokerId) -> true, null));
new DefaultElectPolicy((cluster, brokerName, brokerId) -> true, null));
assertEquals(cResult2.getResponseCode(), ResponseCode.CONTROLLER_MASTER_STILL_EXIST);
// admin successful elect a assignedMaster.
final ElectMasterRequestHeader assignRequest2 = ElectMasterRequestHeader.ofAdminTrigger(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, 1L);
final ControllerResult<ElectMasterResponseHeader> cResult3 = this.replicasInfoManager.electMaster(assignRequest2,
new DefaultElectPolicy((cluster, brokerName, brokerId) -> !brokerId.equals(response.getMasterBrokerId()), null));
new DefaultElectPolicy((cluster, brokerName, brokerId) -> !brokerId.equals(response.getMasterBrokerId()), null));
assertEquals(cResult3.getResponseCode(), ResponseCode.SUCCESS);
final ElectMasterResponseHeader response3 = cResult3.getResponse();
@@ -416,7 +421,7 @@ public class ReplicasInfoManagerTest {
// However, the syncStateSet in statemachine is {DEFAULT_IP[0]}, not more replicas can be elected as master, it will be failed.
final ElectMasterRequestHeader electRequest = ElectMasterRequestHeader.ofControllerTrigger(DEFAULT_BROKER_NAME);
final ControllerResult<ElectMasterResponseHeader> cResult = this.replicasInfoManager.electMaster(electRequest,
new DefaultElectPolicy((cluster, brokerName, brokerId) -> !brokerId.equals(1L), null));
new DefaultElectPolicy((cluster, brokerName, brokerId) -> !brokerId.equals(1L), null));
final List<EventMessage> events = cResult.getEvents();
assertEquals(events.size(), 1);
final ElectMasterEvent event = (ElectMasterEvent) events.get(0);
@@ -463,4 +468,44 @@ public class ReplicasInfoManagerTest {
assertEquals(ResponseCode.SUCCESS, result7.getResponseCode());
}
@Test
public void testSerialize() {
mockMetaData();
byte[] data;
try {
data = this.replicasInfoManager.serialize();
} catch (Throwable e) {
throw new RuntimeException(e);
}
final ReplicasInfoManager newReplicasInfoManager = new ReplicasInfoManager(config);
try {
newReplicasInfoManager.deserializeFrom(data);
} catch (Throwable e) {
throw new RuntimeException(e);
}
Map<String, BrokerReplicaInfo> oldReplicaInfoTable = new TreeMap<>();
Map<String, BrokerReplicaInfo> newReplicaInfoTable = new TreeMap<>();
Map<String/* brokerName */, SyncStateInfo> oldSyncStateTable = new TreeMap<>();
Map<String/* brokerName */, SyncStateInfo> newSyncStateTable = new TreeMap<>();
try {
Field field = ReplicasInfoManager.class.getDeclaredField("replicaInfoTable");
field.setAccessible(true);
oldReplicaInfoTable.putAll((Map<String, BrokerReplicaInfo>) field.get(this.replicasInfoManager));
newReplicaInfoTable.putAll((Map<String, BrokerReplicaInfo>) field.get(newReplicasInfoManager));
field = ReplicasInfoManager.class.getDeclaredField("syncStateSetInfoTable");
field.setAccessible(true);
oldSyncStateTable.putAll((Map<String, SyncStateInfo>) field.get(this.replicasInfoManager));
newSyncStateTable.putAll((Map<String, SyncStateInfo>) field.get(newReplicasInfoManager));
} catch (NoSuchFieldException | IllegalAccessException e) {
throw new RuntimeException(e);
}
assertArrayEquals(oldReplicaInfoTable.keySet().toArray(), newReplicaInfoTable.keySet().toArray());
assertArrayEquals(oldSyncStateTable.keySet().toArray(), newSyncStateTable.keySet().toArray());
for (String brokerName : oldReplicaInfoTable.keySet()) {
BrokerReplicaInfo oldReplicaInfo = oldReplicaInfoTable.get(brokerName);
BrokerReplicaInfo newReplicaInfo = newReplicaInfoTable.get(brokerName);
Field[] fields = oldReplicaInfo.getClass().getFields();
}
}
}