opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

Polish the code structure and code style

Browse Source
This commit is contained in:
RongtongJin
2023-03-11 12:54:40 +08:00
committed by rongtong
parent 578a42803c
commit 28618e95ab
13 changed files with 36 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java
+1
View File
@@ -17,6 +17,7 @@
package org.apache.rocketmq.controller;
import io.netty.channel.Channel;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
public interface BrokerHeartbeatManager {
controller/src/main/java/org/apache/rocketmq/controller/ControllerManager.java
+1 -1
View File
@@ -33,7 +33,7 @@ import org.apache.rocketmq.common.future.FutureTaskExt;
import org.apache.rocketmq.controller.elect.impl.DefaultElectPolicy;
import org.apache.rocketmq.controller.impl.DLedgerController;
import org.apache.rocketmq.controller.impl.DefaultBrokerHeartbeatManager;
import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
import org.apache.rocketmq.controller.processor.ControllerRequestProcessor;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
controller/src/main/java/org/apache/rocketmq/controller/elect/impl/DefaultElectPolicy.java
+1 -1
View File
@@ -17,7 +17,7 @@
package org.apache.rocketmq.controller.elect.impl;
import org.apache.rocketmq.controller.elect.ElectPolicy;
import org.apache.rocketmq.controller.BrokerLiveInfo;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
import org.apache.rocketmq.controller.helper.BrokerLiveInfoGetter;
import org.apache.rocketmq.controller.helper.BrokerValidPredicate;
controller/src/main/java/org/apache/rocketmq/controller/helper/BrokerLiveInfoGetter.java
+1 -1
View File
@@ -17,7 +17,7 @@
package org.apache.rocketmq.controller.helper;
import org.apache.rocketmq.controller.BrokerLiveInfo;
import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
public interface BrokerLiveInfoGetter {
common/src/main/java/org/apache/rocketmq/common/BrokerAddrInfo.java → controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/BrokerIdentityInfo.java
+6 -6
View File
@@ -14,18 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.common;
package org.apache.rocketmq.controller.impl.heartbeat;
import java.util.Objects;
public class BrokerAddrInfo {
public class BrokerIdentityInfo {
private final String clusterName;
private final String brokerName;
private final Long brokerId;
public BrokerAddrInfo(String clusterName, String brokerName, Long brokerId) {
public BrokerIdentityInfo(String clusterName, String brokerName, Long brokerId) {
this.clusterName = clusterName;
this.brokerName = brokerName;
this.brokerId = brokerId;
@@ -56,8 +56,8 @@ public class BrokerAddrInfo {
return false;
}
if (obj instanceof BrokerAddrInfo) {
BrokerAddrInfo addr = (BrokerAddrInfo) obj;
if (obj instanceof BrokerIdentityInfo) {
BrokerIdentityInfo addr = (BrokerIdentityInfo) obj;
return clusterName.equals(addr.clusterName) && brokerName.equals(addr.brokerName) && brokerId.equals(addr.brokerId);
}
return false;
@@ -70,7 +70,7 @@ public class BrokerAddrInfo {
@Override
public String toString() {
return "BrokerAddrInfo{" +
return "BrokerIdentityInfo{" +
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerId=" + brokerId +
controller/src/main/java/org/apache/rocketmq/controller/BrokerLiveInfo.java → controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/BrokerLiveInfo.java
+1 -1
View File
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.controller;
package org.apache.rocketmq.controller.impl.heartbeat;
import io.netty.channel.Channel;
controller/src/main/java/org/apache/rocketmq/controller/impl/DefaultBrokerHeartbeatManager.java → controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java
+12 -14
View File
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.controller.impl;
package org.apache.rocketmq.controller.impl.heartbeat;
import io.netty.channel.Channel;
import java.util.ArrayList;
@@ -27,12 +27,10 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.BrokerAddrInfo;
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.BrokerLiveInfo;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
@@ -44,7 +42,7 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
private final ExecutorService executor = Executors.newFixedThreadPool(2, new ThreadFactoryImpl("DefaultBrokerHeartbeatManager_executorService_"));
private final ControllerConfig controllerConfig;
private final Map<BrokerAddrInfo/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
private final Map<BrokerIdentityInfo/* brokerIdentity*/, BrokerLiveInfo> brokerLiveTable;
private final List<BrokerLifecycleListener> brokerLifecycleListeners;
public DefaultBrokerHeartbeatManager(final ControllerConfig controllerConfig) {
@@ -67,9 +65,9 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
public void scanNotActiveBroker() {
try {
log.info("start scanNotActiveBroker");
final Iterator<Map.Entry<BrokerAddrInfo, BrokerLiveInfo>> iterator = this.brokerLiveTable.entrySet().iterator();
final Iterator<Map.Entry<BrokerIdentityInfo, BrokerLiveInfo>> iterator = this.brokerLiveTable.entrySet().iterator();
while (iterator.hasNext()) {
final Map.Entry<BrokerAddrInfo, BrokerLiveInfo> next = iterator.next();
final Map.Entry<BrokerIdentityInfo, BrokerLiveInfo> next = iterator.next();
long last = next.getValue().getLastUpdateTimestamp();
long timeoutMillis = next.getValue().getHeartbeatTimeoutMillis();
if (System.currentTimeMillis() - last > timeoutMillis) {
@@ -102,8 +100,8 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
@Override
public void onBrokerHeartbeat(String clusterName, String brokerName, String brokerAddr, Long brokerId,
Long timeoutMillis, Channel channel, Integer epoch, Long maxOffset, Long confirmOffset, Integer electionPriority) {
BrokerAddrInfo addrInfo = new BrokerAddrInfo(clusterName, brokerName, brokerId);
BrokerLiveInfo prev = this.brokerLiveTable.get(addrInfo);
BrokerIdentityInfo brokerIdentityInfo = new BrokerIdentityInfo(clusterName, brokerName, brokerId);
BrokerLiveInfo prev = this.brokerLiveTable.get(brokerIdentityInfo);
int realEpoch = Optional.ofNullable(epoch).orElse(-1);
long realBrokerId = Optional.ofNullable(brokerId).orElse(-1L);
long realMaxOffset = Optional.ofNullable(maxOffset).orElse(-1L);
@@ -111,7 +109,7 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
long realTimeoutMillis = Optional.ofNullable(timeoutMillis).orElse(DEFAULT_BROKER_CHANNEL_EXPIRED_TIME);
int realElectionPriority = Optional.ofNullable(electionPriority).orElse(Integer.MAX_VALUE);
if (null == prev) {
this.brokerLiveTable.put(addrInfo,
this.brokerLiveTable.put(brokerIdentityInfo,
new BrokerLiveInfo(brokerName,
brokerAddr,
realBrokerId,
@@ -121,7 +119,7 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
realEpoch,
realMaxOffset,
realElectionPriority));
log.info("new broker registered, {}, brokerId:{}", addrInfo, realBrokerId);
log.info("new broker registered, {}, brokerId:{}", brokerIdentityInfo, realBrokerId);
} else {
prev.setLastUpdateTimestamp(System.currentTimeMillis());
prev.setHeartbeatTimeoutMillis(realTimeoutMillis);
@@ -137,8 +135,8 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
@Override
public void onBrokerChannelClose(Channel channel) {
BrokerAddrInfo addrInfo = null;
for (Map.Entry<BrokerAddrInfo, BrokerLiveInfo> entry : this.brokerLiveTable.entrySet()) {
BrokerIdentityInfo addrInfo = null;
for (Map.Entry<BrokerIdentityInfo, BrokerLiveInfo> entry : this.brokerLiveTable.entrySet()) {
if (entry.getValue().getChannel() == channel) {
log.info("Channel {} inactive, broker {}, addr:{}, id:{}", entry.getValue().getChannel(), entry.getValue().getBrokerName(), entry.getValue().getBrokerAddr(), entry.getValue().getBrokerId());
addrInfo = entry.getKey();
@@ -154,12 +152,12 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
@Override
public BrokerLiveInfo getBrokerLiveInfo(String clusterName, String brokerName, Long brokerId) {
return this.brokerLiveTable.get(new BrokerAddrInfo(clusterName, brokerName, brokerId));
return this.brokerLiveTable.get(new BrokerIdentityInfo(clusterName, brokerName, brokerId));
}
@Override
public boolean isBrokerActive(String clusterName, String brokerName, Long brokerId) {
final BrokerLiveInfo info = this.brokerLiveTable.get(new BrokerAddrInfo(clusterName, brokerName, brokerId));
final BrokerLiveInfo info = this.brokerLiveTable.get(new BrokerIdentityInfo(clusterName, brokerName, brokerId));
if (info != null) {
long last = info.getLastUpdateTimestamp();
long timeoutMillis = info.getHeartbeatTimeoutMillis();
controller/src/main/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManager.java
+2 -2
View File
@@ -187,8 +187,8 @@ public class ReplicasInfoManager {
// elect by policy
if (newMaster == null) {
// we should assign this assignedBrokerAddr when the brokerAddress need to be elected by force
Long assignedBrokerId = request.isForceElect() ? brokerId : null;
// 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);
}
controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/DefaultBrokerHeartbeatManagerTest.java
+1 -1
View File
@@ -20,7 +20,7 @@ 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.DefaultBrokerHeartbeatManager;
import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
import org.junit.Before;
import org.junit.Test;
controller/src/test/java/org/apache/rocketmq/controller/impl/controller/impl/manager/ReplicasInfoManagerTest.java
+1 -1
View File
@@ -25,7 +25,7 @@ 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.DefaultBrokerHeartbeatManager;
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;
namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
+1 -1
View File
@@ -1129,7 +1129,7 @@ class BrokerAddrInfo {
@Override
public String toString() {
return "BrokerAddrInfo [clusterName=" + clusterName + ", brokerAddr=" + brokerAddr + "]";
return "BrokerIdentityInfo [clusterName=" + clusterName + ", brokerAddr=" + brokerAddr + "]";
}
}
remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/controller/ElectMasterRequestHeader.java
+7 -25
View File
@@ -32,14 +32,14 @@ public class ElectMasterRequestHeader implements CommandCustomHeader {
* brokerId
* for brokerTrigger electMaster: this brokerId will be elected as a master when it is the first time to elect
* in this broker-set
* for adminTrigger electMaster: this brokerAddress is also named assignedBrokerId, which means we must prefer to elect
* for adminTrigger electMaster: this brokerId is also named assignedBrokerId, which means we must prefer to elect
* it as a new master when this broker is valid.
*/
@CFNotNull
private Long brokerId;
@CFNotNull
private Boolean forceElect = false;
private Boolean designateElect = false;
public ElectMasterRequestHeader() {
}
@@ -54,11 +54,11 @@ public class ElectMasterRequestHeader implements CommandCustomHeader {
this.brokerId = brokerId;
}
public ElectMasterRequestHeader(String clusterName, String brokerName, Long brokerId, boolean forceElect) {
public ElectMasterRequestHeader(String clusterName, String brokerName, Long brokerId, boolean designateElect) {
this.clusterName = clusterName;
this.brokerName = brokerName;
this.brokerId = brokerId;
this.forceElect = forceElect;
this.designateElect = designateElect;
}
public static ElectMasterRequestHeader ofBrokerTrigger(String clusterName, String brokerName,
@@ -98,8 +98,8 @@ public class ElectMasterRequestHeader implements CommandCustomHeader {
this.clusterName = clusterName;
}
public boolean isForceElect() {
return this.forceElect;
public boolean getDesignateElect() {
return this.designateElect;
}
@Override
@@ -108,29 +108,11 @@ public class ElectMasterRequestHeader implements CommandCustomHeader {
"clusterName='" + clusterName + '\'' +
", brokerName='" + brokerName + '\'' +
", brokerId=" + brokerId +
", forceElect=" + forceElect +
", designateElect=" + designateElect +
'}';
}
@Override
public void checkFields() throws RemotingCommandException {
}
/**
* The elect master request's type
*/
public enum ElectMasterTriggerType {
/**
* Trigger by broker
*/
BROKER_TRIGGER,
/**
* Trigger by controller
*/
CONTROLLER_TRIGGER,
/**
* Trigger by admin
*/
ADMIN_TRIGGER
}
}
tools/src/main/java/org/apache/rocketmq/tools/command/controller/ReElectMasterSubCommand.java
+1 -1
View File
@@ -50,7 +50,7 @@ public class ReElectMasterSubCommand implements SubCommand {
opt.setRequired(true);
options.addOption(opt);
opt = new Option("n", "brokerName", true, "The broker name of the replicas that require to be manipulated");
opt = new Option("bn", "brokerName", true, "The broker name of the replicas that require to be manipulated");
opt.setRequired(true);
options.addOption(opt);