database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[enhancement](k8s) Support fqdn mode for fe in k8s enviroment (#17329)

Browse Source
This commit is contained in:
yinzhijian
2023-03-05 10:18:56 +08:00
committed by GitHub
parent 7b4fc412c5
commit 627b5ee302
29 changed files with 563 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
fe/fe-core/src/main/java/org/apache/doris/alter/SystemHandler.java
View File

@ -153,16 +153,20 @@ public class SystemHandler extends AlterHandler {
} else if (alterClause instanceof AddObserverClause) {
AddObserverClause clause = (AddObserverClause) alterClause;
Env.getCurrentEnv().addFrontend(FrontendNodeType.OBSERVER, clause.getHost(), clause.getPort());
Env.getCurrentEnv().addFrontend(FrontendNodeType.OBSERVER, clause.getIp(), clause.getHostName(),
clause.getPort());
} else if (alterClause instanceof DropObserverClause) {
DropObserverClause clause = (DropObserverClause) alterClause;
Env.getCurrentEnv().dropFrontend(FrontendNodeType.OBSERVER, clause.getHost(), clause.getPort());
Env.getCurrentEnv().dropFrontend(FrontendNodeType.OBSERVER, clause.getIp(), clause.getHostName(),
clause.getPort());
} else if (alterClause instanceof AddFollowerClause) {
AddFollowerClause clause = (AddFollowerClause) alterClause;
Env.getCurrentEnv().addFrontend(FrontendNodeType.FOLLOWER, clause.getHost(), clause.getPort());
Env.getCurrentEnv().addFrontend(FrontendNodeType.FOLLOWER, clause.getIp(), clause.getHostName(),
clause.getPort());
} else if (alterClause instanceof DropFollowerClause) {
DropFollowerClause clause = (DropFollowerClause) alterClause;
Env.getCurrentEnv().dropFrontend(FrontendNodeType.FOLLOWER, clause.getHost(), clause.getPort());
Env.getCurrentEnv().dropFrontend(FrontendNodeType.FOLLOWER, clause.getIp(), clause.getHostName(),
clause.getPort());
} else if (alterClause instanceof ModifyBrokerClause) {
ModifyBrokerClause clause = (ModifyBrokerClause) alterClause;
Env.getCurrentEnv().getBrokerMgr().execute(clause);

22
fe/fe-core/src/main/java/org/apache/doris/analysis/FrontendClause.java
View File

@ -22,11 +22,11 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.Pair;
import org.apache.doris.ha.FrontendNodeType;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.system.SystemInfoService;
import org.apache.doris.system.SystemInfoService.HostInfo;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
@ -36,7 +36,8 @@ import java.util.Map;
public class FrontendClause extends AlterClause {
protected String hostPort;
protected String host;
protected String ip;
protected String hostName;
protected int port;
protected FrontendNodeType role;
@ -46,8 +47,12 @@ public class FrontendClause extends AlterClause {
this.role = role;
}
public String getHost() {
return host;
public String getIp() {
return ip;
}
public String getHostName() {
return hostName;
}
public int getPort() {
@ -61,10 +66,11 @@ public class FrontendClause extends AlterClause {
analyzer.getQualifiedUser());
}
Pair<String, Integer> pair = SystemInfoService.validateHostAndPort(hostPort);
this.host = pair.first;
this.port = pair.second;
Preconditions.checkState(!Strings.isNullOrEmpty(host));
HostInfo hostInfo = SystemInfoService.getIpHostAndPort(hostPort, true);
this.ip = hostInfo.getIp();
this.hostName = hostInfo.getHostName();
this.port = hostInfo.getPort();
Preconditions.checkState(!Strings.isNullOrEmpty(ip));
}
@Override

280
fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
View File

@ -220,6 +220,7 @@ import org.apache.doris.system.FQDNManager;
import org.apache.doris.system.Frontend;
import org.apache.doris.system.HeartbeatMgr;
import org.apache.doris.system.SystemInfoService;
import org.apache.doris.system.SystemInfoService.HostInfo;
import org.apache.doris.task.AgentBatchTask;
import org.apache.doris.task.AgentTaskExecutor;
import org.apache.doris.task.CompactionTask;
@ -351,9 +352,7 @@ public class Env {
private FrontendNodeType feType;
// replica and observer use this value to decide provide read service or not
private long synchronizedTimeMs;
private int masterRpcPort;
private int masterHttpPort;
private String masterIp;
private MasterInfo masterInfo;
private MetaIdGenerator idGenerator = new MetaIdGenerator(NEXT_ID_INIT_VALUE);
@ -366,8 +365,8 @@ public class Env {
private static Env CHECKPOINT = null;
private static long checkpointThreadId = -1;
private Checkpoint checkpointer;
private List<Pair<String, Integer>> helperNodes = Lists.newArrayList();
private Pair<String, Integer> selfNode = null;
private List<HostInfo> helperNodes = Lists.newArrayList();
private HostInfo selfNode = null;
// node name -> Frontend
private ConcurrentHashMap<String, Frontend> frontends;
@ -576,9 +575,7 @@ public class Env {
this.removedFrontends = new ConcurrentLinkedQueue<>();
this.journalObservable = new JournalObservable();
this.masterRpcPort = 0;
this.masterHttpPort = 0;
this.masterIp = "";
this.masterInfo = new MasterInfo();
this.systemInfo = new SystemInfoService();
this.heartbeatMgr = new HeartbeatMgr(systemInfo, !isCheckpointCatalog);
@ -926,7 +923,8 @@ public class Env {
// For compatibility. Because this is the very first time to start, so we arbitrarily choose
// a new name for this node
role = FrontendNodeType.FOLLOWER;
nodeName = genFeNodeName(selfNode.first, selfNode.second, false /* new style */);
nodeName = genFeNodeName(Config.enable_fqdn_mode ? selfNode.getIdent() : selfNode.getIp(),
selfNode.getPort(), false /* new style */);
storage.writeFrontendRoleAndNodeName(role, nodeName);
LOG.info("very first time to start this node. role: {}, node name: {}", role.name(), nodeName);
} else {
@ -942,24 +940,13 @@ public class Env {
// But we will get a empty nodeName after upgrading.
// So for forward compatibility, we use the "old-style" way of naming: "ip_port",
// and update the ROLE file.
nodeName = genFeNodeName(selfNode.first, selfNode.second, true/* old style */);
nodeName = genFeNodeName(selfNode.getIp(), selfNode.getPort(), true/* old style */);
storage.writeFrontendRoleAndNodeName(role, nodeName);
LOG.info("forward compatibility. role: {}, node name: {}", role.name(), nodeName);
} else {
// nodeName should be like "192.168.1.1_9217_1620296111213"
// and the selfNode should be the prefix of nodeName.
// If not, it means that the ip used last time is different from this time, which is not allowed.
// But is metadata_failure_recovery is true,
// we will not check it because this may be a FE migration.
String[] split = nodeName.split("_");
if (Config.metadata_failure_recovery.equals("false") && !selfNode.first.equalsIgnoreCase(
split[0])) {
throw new IOException(
"the self host " + selfNode.first + " does not equal to the host in ROLE" + " file "
+ split[0] + ". You need to set 'priority_networks' config"
+ " in fe.conf to match the host " + split[0]);
}
}
// Notice:
// With the introduction of FQDN, the nodeName is no longer bound to an IP address,
// so consistency is no longer checked here. Otherwise, the startup will fail.
}
Preconditions.checkNotNull(role);
@ -972,7 +959,8 @@ public class Env {
storage.writeClusterIdAndToken();
isFirstTimeStartUp = true;
Frontend self = new Frontend(role, nodeName, selfNode.first, selfNode.second);
Frontend self = new Frontend(role, nodeName, selfNode.getIp(), selfNode.getHostName(),
selfNode.getPort());
// We don't need to check if frontends already contains self.
// frontends must be empty cause no image is loaded and no journal is replayed yet.
// And this frontend will be persisted later after opening bdbje environment.
@ -1016,7 +1004,7 @@ public class Env {
Preconditions.checkNotNull(role);
Preconditions.checkNotNull(nodeName);
Pair<String, Integer> rightHelperNode = helperNodes.get(0);
HostInfo rightHelperNode = helperNodes.get(0);
Storage storage = new Storage(this.imageDir);
if (roleFile.exists() && (role != storage.getRole() || !nodeName.equals(storage.getNodeName()))
@ -1026,8 +1014,8 @@ public class Env {
if (!versionFile.exists()) {
// If the version file doesn't exist, download it from helper node
if (!getVersionFileFromHelper(rightHelperNode)) {
throw new IOException(
"fail to download version file from " + rightHelperNode.first + " will exit.");
throw new IOException("fail to download version file from "
+ rightHelperNode.getIp() + " will exit.");
}
// NOTE: cluster_id will be init when Storage object is constructed,
@ -1044,7 +1032,7 @@ public class Env {
clusterId = storage.getClusterID();
token = storage.getToken();
try {
URL idURL = new URL("http://" + NetUtils.getHostPortInAccessibleFormat(rightHelperNode.first, Config.http_port) + "/check");
URL idURL = new URL("http://" + NetUtils.getHostPortInAccessibleFormat(rightHelperNode.getIp(), Config.http_port) + "/check");
HttpURLConnection conn = null;
conn = (HttpURLConnection) idURL.openConnection();
conn.setConnectTimeout(2 * 1000);
@ -1052,8 +1040,11 @@ public class Env {
String clusterIdString = conn.getHeaderField(MetaBaseAction.CLUSTER_ID);
int remoteClusterId = Integer.parseInt(clusterIdString);
if (remoteClusterId != clusterId) {
LOG.error("cluster id is not equal with helper node {}. will exit.", rightHelperNode.first);
System.exit(-1);
LOG.error("cluster id is not equal with helper node {}. will exit.",
rightHelperNode.getIp());
throw new IOException(
"cluster id is not equal with helper node "
+ rightHelperNode.getIp() + ". will exit.");
}
String remoteToken = conn.getHeaderField(MetaBaseAction.TOKEN);
if (token == null && remoteToken != null) {
@ -1067,7 +1058,8 @@ public class Env {
Preconditions.checkNotNull(remoteToken);
if (!token.equals(remoteToken)) {
throw new IOException(
"token is not equal with helper node " + rightHelperNode.first + ". will exit.");
"token is not equal with helper node "
+ rightHelperNode.getIp() + ". will exit.");
}
}
} catch (Exception e) {
@ -1089,7 +1081,8 @@ public class Env {
}
Preconditions.checkState(helperNodes.size() == 1);
LOG.info("finished to get cluster id: {}, role: {} and node name: {}", clusterId, role.name(), nodeName);
LOG.info("finished to get cluster id: {}, isElectable: {}, role: {} and node name: {}",
clusterId, isElectable, role.name(), nodeName);
}
public static String genFeNodeName(String host, int port, boolean isOldStyle) {
@ -1106,12 +1099,14 @@ public class Env {
private boolean getFeNodeTypeAndNameFromHelpers() {
// we try to get info from helper nodes, once we get the right helper node,
// other helper nodes will be ignored and removed.
Pair<String, Integer> rightHelperNode = null;
for (Pair<String, Integer> helperNode : helperNodes) {
HostInfo rightHelperNode = null;
for (HostInfo helperNode : helperNodes) {
try {
URL url = new URL("http://" + NetUtils.getHostPortInAccessibleFormat(helperNode.first, Config.http_port)
+ "/role?host=" + selfNode.first
+ "&port=" + selfNode.second);
// For upgrade compatibility, the host parameter name remains the same
// and the new hostname parameter is added
URL url = new URL("http://" + NetUtils.getHostPortInAccessibleFormat(helperNode.getIp(), Config.http_port)
+ "/role?host=" + selfNode.getIp() + "&hostname=" + selfNode.getHostName()
+ "&port=" + selfNode.getPort());
HttpURLConnection conn = null;
conn = (HttpURLConnection) url.openConnection();
if (conn.getResponseCode() != 200) {
@ -1138,14 +1133,15 @@ public class Env {
if (Strings.isNullOrEmpty(nodeName)) {
// For forward compatibility, we use old-style name: "ip_port"
nodeName = genFeNodeName(selfNode.first, selfNode.second, true /* old style */);
nodeName = genFeNodeName(selfNode.getIp(), selfNode.getPort(), true /* old style */);
}
} catch (Exception e) {
LOG.warn("failed to get fe node type from helper node: {}.", helperNode, e);
continue;
}
LOG.info("get fe node type {}, name {} from {}:{}", role, nodeName, helperNode.first, Config.http_port);
LOG.info("get fe node type {}, name {} from {}:{}:{}", role, nodeName,
helperNode.getHostName(), helperNode.getIp(), Config.http_port);
rightHelperNode = helperNode;
break;
}
@ -1160,7 +1156,9 @@ public class Env {
}
private void getSelfHostPort() {
selfNode = Pair.of(FrontendOptions.getLocalHostAddress(), Config.edit_log_port);
String hostName = Strings.nullToEmpty(FrontendOptions.getHostName());
selfNode = new HostInfo(FrontendOptions.getLocalHostAddress(), hostName,
Config.edit_log_port);
LOG.debug("get self node: {}", selfNode);
}
@ -1193,8 +1191,8 @@ public class Env {
if (helpers != null) {
String[] splittedHelpers = helpers.split(",");
for (String helper : splittedHelpers) {
Pair<String, Integer> helperHostPort = SystemInfoService.validateHostAndPort(helper);
if (helperHostPort.equals(selfNode)) {
HostInfo helperHostPort = SystemInfoService.getIpHostAndPort(helper, true);
if (helperHostPort.isSame(selfNode)) {
/**
* If user specified the helper node to this FE itself,
* we will stop the starting FE process and report an error.
@ -1211,7 +1209,7 @@ public class Env {
}
} else {
// If helper node is not designated, use local node as helper node.
helperNodes.add(Pair.of(selfNode.first, Config.edit_log_port));
helperNodes.add(new HostInfo(selfNode.getIp(), selfNode.getHostName(), Config.edit_log_port));
}
}
@ -1234,7 +1232,8 @@ public class Env {
// This is not the first time this node start up.
// It should already added to FE group, just set helper node as it self.
LOG.info("role file exist. this is not the first time to start up");
helperNodes = Lists.newArrayList(Pair.of(selfNode.first, Config.edit_log_port));
helperNodes = Lists.newArrayList(new HostInfo(selfNode.getIp(), selfNode.getHostName(),
Config.edit_log_port));
return;
}
@ -1309,11 +1308,11 @@ public class Env {
// MUST set master ip before starting checkpoint thread.
// because checkpoint thread need this info to select non-master FE to push image
this.masterIp = FrontendOptions.getLocalHostAddress();
this.masterRpcPort = Config.rpc_port;
this.masterHttpPort = Config.http_port;
MasterInfo info = new MasterInfo(this.masterIp, this.masterHttpPort, this.masterRpcPort);
editLog.logMasterInfo(info);
this.masterInfo = new MasterInfo(Env.getCurrentEnv().getSelfNode().getIp(),
Env.getCurrentEnv().getSelfNode().getHostName(),
Config.http_port,
Config.rpc_port);
editLog.logMasterInfo(masterInfo);
// for master, the 'isReady' is set behind.
// but we are sure that all metadata is replayed if we get here.
@ -1461,7 +1460,7 @@ public class Env {
if (Config.edit_log_type.equalsIgnoreCase("bdb")) {
for (Frontend fe : frontends.values()) {
if (fe.getRole() == FrontendNodeType.FOLLOWER || fe.getRole() == FrontendNodeType.REPLICA) {
((BDBHA) getHaProtocol()).addHelperSocket(fe.getHost(), fe.getEditLogPort());
((BDBHA) getHaProtocol()).addHelperSocket(fe.getIp(), fe.getEditLogPort());
}
}
}
@ -1531,11 +1530,11 @@ public class Env {
return;
}
Frontend fe = checkFeExist(selfNode.first, selfNode.second);
Frontend fe = checkFeExist(selfNode.getIp(), selfNode.getHostName(), selfNode.getPort());
if (fe == null) {
LOG.error("current node {}:{} is not added to the cluster, will exit."
LOG.error("current node {}:{}:{} is not added to the cluster, will exit."
+ " Your FE IP maybe changed, please set 'priority_networks' config in fe.conf properly.",
selfNode.first, selfNode.second);
selfNode.getHostName(), selfNode.getIp(), selfNode.getPort());
System.exit(-1);
} else if (fe.getRole() != role) {
LOG.error("current node role is {} not match with frontend recorded role {}. will exit", role,
@ -1553,9 +1552,9 @@ public class Env {
}
}
private boolean getVersionFileFromHelper(Pair<String, Integer> helperNode) throws IOException {
private boolean getVersionFileFromHelper(HostInfo helperNode) throws IOException {
try {
String url = "http://" + NetUtils.getHostPortInAccessibleFormat(helperNode.first, Config.http_port) + "/version";
String url = "http://" + NetUtils.getHostPortInAccessibleFormat(helperNode.getIp(), Config.http_port) + "/version";
File dir = new File(this.imageDir);
MetaHelper.getRemoteFile(url, HTTP_TIMEOUT_SECOND * 1000,
MetaHelper.getOutputStream(Storage.VERSION_FILE, dir));
@ -1568,13 +1567,13 @@ public class Env {
return false;
}
private void getNewImage(Pair<String, Integer> helperNode) throws IOException {
private void getNewImage(HostInfo helperNode) throws IOException {
long localImageVersion = 0;
Storage storage = new Storage(this.imageDir);
localImageVersion = storage.getLatestImageSeq();
try {
String hostPort = NetUtils.getHostPortInAccessibleFormat(helperNode.first, Config.http_port);
String hostPort = NetUtils.getHostPortInAccessibleFormat(helperNode.getIp(), Config.http_port);
URL infoUrl = new URL("http://" + hostPort + "/info");
StorageInfo info = getStorageInfo(infoUrl);
long version = info.getImageSeq();
@ -1597,10 +1596,11 @@ public class Env {
Preconditions.checkNotNull(selfNode);
Preconditions.checkNotNull(helperNodes);
LOG.debug("self: {}. helpers: {}", selfNode, helperNodes);
// if helper nodes contain it self, remove other helpers
// if helper nodes contain itself, remove other helpers
boolean containSelf = false;
for (Pair<String, Integer> helperNode : helperNodes) {
if (selfNode.equals(helperNode)) {
for (HostInfo helperNode : helperNodes) {
// WARN: cannot use equals() here, because the hostname may not equal to helperNode.getHostName()
if (selfNode.isSame(helperNode)) {
containSelf = true;
break;
}
@ -1698,11 +1698,9 @@ public class Env {
}
public long loadMasterInfo(DataInputStream dis, long checksum) throws IOException {
masterIp = Text.readString(dis);
masterRpcPort = dis.readInt();
long newChecksum = checksum ^ masterRpcPort;
masterHttpPort = dis.readInt();
newChecksum ^= masterHttpPort;
masterInfo = MasterInfo.read(dis);
long newChecksum = checksum ^ masterInfo.getRpcPort();
newChecksum ^= masterInfo.getHttpPort();
LOG.info("finished replay masterInfo from image");
return newChecksum;
@ -1985,14 +1983,9 @@ public class Env {
}
public long saveMasterInfo(CountingDataOutputStream dos, long checksum) throws IOException {
Text.writeString(dos, masterIp);
checksum ^= masterRpcPort;
dos.writeInt(masterRpcPort);
checksum ^= masterHttpPort;
dos.writeInt(masterHttpPort);
masterInfo.write(dos);
checksum ^= masterInfo.getRpcPort();
checksum ^= masterInfo.getHttpPort();
return checksum;
}
@ -2452,58 +2445,100 @@ public class Env {
};
}
public void addFrontend(FrontendNodeType role, String host, int editLogPort) throws DdlException {
public void addFrontend(FrontendNodeType role, String ip, String hostname, int editLogPort) throws DdlException {
if (!tryLock(false)) {
throw new DdlException("Failed to acquire catalog lock. Try again");
}
try {
Frontend fe = checkFeExist(host, editLogPort);
Frontend fe = checkFeExist(ip, hostname, editLogPort);
if (fe != null) {
throw new DdlException("frontend already exists " + fe);
}
if (Config.enable_fqdn_mode && StringUtils.isEmpty(hostname)) {
throw new DdlException("frontend's hostName should not be empty while enable_fqdn_mode is true");
}
String host = hostname != null && Config.enable_fqdn_mode ? hostname : ip;
String nodeName = genFeNodeName(host, editLogPort, false /* new name style */);
if (removedFrontends.contains(nodeName)) {
throw new DdlException("frontend name already exists " + nodeName + ". Try again");
}
fe = new Frontend(role, nodeName, host, editLogPort);
fe = new Frontend(role, nodeName, ip, hostname, editLogPort);
frontends.put(nodeName, fe);
BDBHA bdbha = (BDBHA) haProtocol;
if (role == FrontendNodeType.FOLLOWER || role == FrontendNodeType.REPLICA) {
bdbha.addHelperSocket(host, editLogPort);
helperNodes.add(Pair.of(host, editLogPort));
bdbha.addHelperSocket(ip, editLogPort);
helperNodes.add(new HostInfo(ip, hostname, editLogPort));
bdbha.addUnReadyElectableNode(nodeName, getFollowerCount());
}
bdbha.removeConflictNodeIfExist(host, editLogPort);
bdbha.removeConflictNodeIfExist(ip, editLogPort);
editLog.logAddFrontend(fe);
} finally {
unlock();
}
}
public void dropFrontend(FrontendNodeType role, String host, int port) throws DdlException {
if (host.equals(selfNode.first) && port == selfNode.second && feType == FrontendNodeType.MASTER) {
public void modifyFrontendIp(String nodeName, String destIp) throws DdlException {
modifyFrontendHost(nodeName, destIp, "");
}
public void modifyFrontendHostName(String nodeName, String destHostName) throws DdlException {
modifyFrontendHost(nodeName, "", destHostName);
}
public void modifyFrontendHost(String nodeName, String destIp, String destHostName) throws DdlException {
if (!tryLock(false)) {
throw new DdlException("Failed to acquire catalog lock. Try again");
}
try {
Frontend fe = getFeByName(nodeName);
if (fe == null) {
throw new DdlException("frontend does not exist, nodeName:" + nodeName);
}
boolean needLog = false;
// we use hostname as address of bdbha, so we don't need to update node address when ip changed
if (!Strings.isNullOrEmpty(destIp) && !destIp.equals(fe.getIp())) {
fe.setIp(destIp);
needLog = true;
}
if (!Strings.isNullOrEmpty(destHostName) && !destHostName.equals(fe.getHostName())) {
fe.setHostName(destHostName);
BDBHA bdbha = (BDBHA) haProtocol;
bdbha.updateNodeAddress(fe.getNodeName(), destHostName, fe.getEditLogPort());
needLog = true;
}
if (needLog) {
Env.getCurrentEnv().getEditLog().logModifyFrontend(fe);
}
} finally {
unlock();
}
}
public void dropFrontend(FrontendNodeType role, String ip, String hostname, int port) throws DdlException {
if (port == selfNode.getPort() && feType == FrontendNodeType.MASTER
&& ((!Strings.isNullOrEmpty(selfNode.getHostName()) && selfNode.getHostName().equals(hostname))
|| ip.equals(selfNode.getIp()))) {
throw new DdlException("can not drop current master node.");
}
if (!tryLock(false)) {
throw new DdlException("Failed to acquire catalog lock. Try again");
}
try {
Frontend fe = checkFeExist(host, port);
Frontend fe = checkFeExist(ip, hostname, port);
if (fe == null) {
throw new DdlException("frontend does not exist[" + host + ":" + port + "]");
throw new DdlException("frontend does not exist[" + ip + ":" + port + "]");
}
if (fe.getRole() != role) {
throw new DdlException(role.toString() + " does not exist[" + host + ":" + port + "]");
throw new DdlException(role.toString() + " does not exist[" + ip + ":" + port + "]");
}
frontends.remove(fe.getNodeName());
removedFrontends.add(fe.getNodeName());
if (fe.getRole() == FrontendNodeType.FOLLOWER || fe.getRole() == FrontendNodeType.REPLICA) {
haProtocol.removeElectableNode(fe.getNodeName());
helperNodes.remove(Pair.of(host, port));
removeHelperNode(ip, hostname, port);
BDBHA ha = (BDBHA) haProtocol;
ha.removeUnReadyElectableNode(nodeName, getFollowerCount());
}
@ -2513,22 +2548,33 @@ public class Env {
}
}
public Frontend checkFeExist(String host, int port) {
private void removeHelperNode(String ip, String hostName, int port) {
// ip may be changed, so we need use both ip and hostname to check.
// use node.getIdent() for simplicity here.
helperNodes.removeIf(node -> (node.getIp().equals(ip)
|| node.getIdent().equals(hostName)) && node.getPort() == port);
}
public Frontend checkFeExist(String ip, String hostName, int port) {
for (Frontend fe : frontends.values()) {
if (fe.getHost().equals(host) && fe.getEditLogPort() == port) {
if (fe.getEditLogPort() != port) {
continue;
}
if (fe.getIp().equals(ip)
|| (!Strings.isNullOrEmpty(fe.getHostName()) && fe.getHostName().equals(hostName))) {
return fe;
}
}
return null;
}
public Frontend getFeByHost(String host) {
public Frontend getFeByIp(String ip) {
for (Frontend fe : frontends.values()) {
InetAddress hostAddr = null;
InetAddress feAddr = null;
try {
hostAddr = InetAddress.getByName(host);
feAddr = InetAddress.getByName(fe.getHost());
hostAddr = InetAddress.getByName(ip);
feAddr = InetAddress.getByName(fe.getIp());
} catch (UnknownHostException e) {
LOG.warn("get address failed: {}", e.getMessage());
return null;
@ -3209,7 +3255,7 @@ public class Env {
public void replayAddFrontend(Frontend fe) {
tryLock(true);
try {
Frontend existFe = checkFeExist(fe.getHost(), fe.getEditLogPort());
Frontend existFe = checkFeExist(fe.getIp(), fe.getHostName(), fe.getEditLogPort());
if (existFe != null) {
LOG.warn("fe {} already exist.", existFe);
if (existFe.getRole() != fe.getRole()) {
@ -3232,13 +3278,34 @@ public class Env {
// DO NOT add helper sockets here, cause BDBHA is not instantiated yet.
// helper sockets will be added after start BDBHA
// But add to helperNodes, just for show
helperNodes.add(Pair.of(fe.getHost(), fe.getEditLogPort()));
helperNodes.add(new HostInfo(fe.getIp(), fe.getHostName(), fe.getEditLogPort()));
}
} finally {
unlock();
}
}
public void replayModifyFrontend(Frontend fe) {
tryLock(true);
try {
Frontend existFe = getFeByName(fe.getNodeName());
if (existFe == null) {
// frontend may already be dropped. this may happen when
// drop and modify operations do not guarantee the order.
return;
}
// modify fe in frontends
existFe.setIp(fe.getIp());
existFe.setHostName(fe.getHostName());
// modify fe in helperNodes
helperNodes.stream().filter(n -> !Strings.isNullOrEmpty(n.getHostName())
&& n.getHostName().equals(fe.getHostName()))
.forEach(n -> n.ip = fe.getIp());
} finally {
unlock();
}
}
public void replayDropFrontend(Frontend frontend) {
tryLock(true);
try {
@ -3248,7 +3315,7 @@ public class Env {
return;
}
if (removedFe.getRole() == FrontendNodeType.FOLLOWER || removedFe.getRole() == FrontendNodeType.REPLICA) {
helperNodes.remove(Pair.of(removedFe.getHost(), removedFe.getEditLogPort()));
removeHelperNode(removedFe.getIp(), removedFe.getHostName(), removedFe.getEditLogPort());
}
removedFrontends.add(removedFe.getNodeName());
@ -3505,16 +3572,16 @@ public class Env {
return this.role;
}
public Pair<String, Integer> getHelperNode() {
public HostInfo getHelperNode() {
Preconditions.checkState(helperNodes.size() >= 1);
return this.helperNodes.get(0);
}
public List<Pair<String, Integer>> getHelperNodes() {
public List<HostInfo> getHelperNodes() {
return Lists.newArrayList(helperNodes);
}
public Pair<String, Integer> getSelfNode() {
public HostInfo getSelfNode() {
return this.selfNode;
}
@ -3530,21 +3597,28 @@ public class Env {
if (!isReady()) {
return 0;
}
return this.masterRpcPort;
return this.masterInfo.getRpcPort();
}
public int getMasterHttpPort() {
if (!isReady()) {
return 0;
}
return this.masterHttpPort;
return this.masterInfo.getHttpPort();
}
public String getMasterIp() {
if (!isReady()) {
return "";
}
return this.masterIp;
return this.masterInfo.getIp();
}
public String getMasterHostName() {
if (!isReady()) {
return "";
}
return this.masterInfo.getHostName();
}
public EsRepository getEsRepository() {
@ -3556,9 +3630,7 @@ public class Env {
}
public void setMaster(MasterInfo info) {
this.masterIp = info.getIp();
this.masterHttpPort = info.getHttpPort();
this.masterRpcPort = info.getRpcPort();
this.masterInfo = info;
}
public boolean canRead() {

47
fe/fe-core/src/main/java/org/apache/doris/common/proc/FrontendsProcNode.java
View File

@ -24,6 +24,7 @@ import org.apache.doris.common.util.NetUtils;
import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.system.Frontend;
import org.apache.doris.system.SystemInfoService.HostInfo;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
@ -31,6 +32,7 @@ import com.google.common.collect.Lists;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
@ -85,12 +87,11 @@ public class FrontendsProcNode implements ProcNodeInterface {
// get all node which are joined in bdb group
List<InetSocketAddress> allFe = env.getHaProtocol().getElectableNodes(true /* include leader */);
allFe.addAll(env.getHaProtocol().getObserverNodes());
List<Pair<String, Integer>> allFeHosts = convertToHostPortPair(allFe);
List<Pair<String, Integer>> helperNodes = env.getHelperNodes();
List<HostInfo> helperNodes = env.getHelperNodes();
// Because the `show frontend` stmt maybe forwarded from other FE.
// if we only get self node from currrent catalog, the "CurrentConnected" field will always points to Msater FE.
String selfNode = Env.getCurrentEnv().getSelfNode().first;
String selfNode = Env.getCurrentEnv().getSelfNode().getIp();
if (ConnectContext.get() != null && !Strings.isNullOrEmpty(ConnectContext.get().getCurrentConnectedFEIp())) {
selfNode = ConnectContext.get().getCurrentConnectedFEIp();
}
@ -99,13 +100,13 @@ public class FrontendsProcNode implements ProcNodeInterface {
List<String> info = new ArrayList<String>();
info.add(fe.getNodeName());
info.add(fe.getHost());
info.add(fe.getIp());
info.add(NetUtils.getHostnameByIp(fe.getHost()));
info.add(NetUtils.getHostnameByIp(fe.getIp()));
info.add(Integer.toString(fe.getEditLogPort()));
info.add(Integer.toString(Config.http_port));
if (fe.getHost().equals(env.getSelfNode().first)) {
if (fe.getIp().equals(env.getSelfNode().getIp())) {
info.add(Integer.toString(Config.query_port));
info.add(Integer.toString(Config.rpc_port));
} else {
@ -114,15 +115,15 @@ public class FrontendsProcNode implements ProcNodeInterface {
}
info.add(fe.getRole().name());
InetSocketAddress socketAddress = new InetSocketAddress(fe.getHost(), fe.getEditLogPort());
InetSocketAddress socketAddress = new InetSocketAddress(fe.getIp(), fe.getEditLogPort());
//An ipv6 address may have different format, so we compare InetSocketAddress objects instead of IP Strings.
//e.g. fdbd:ff1:ce00:1c26::d8 and fdbd:ff1:ce00:1c26:0:0:d8
info.add(String.valueOf(socketAddress.equals(master)));
info.add(Integer.toString(env.getClusterId()));
info.add(String.valueOf(isJoin(allFeHosts, fe)));
info.add(String.valueOf(isJoin(allFe, fe)));
if (fe.getHost().equals(env.getSelfNode().first)) {
if (fe.getIp().equals(env.getSelfNode().getIp())) {
info.add("true");
info.add(Long.toString(env.getEditLog().getMaxJournalId()));
} else {
@ -134,19 +135,35 @@ public class FrontendsProcNode implements ProcNodeInterface {
info.add(fe.getHeartbeatErrMsg());
info.add(fe.getVersion());
// To indicate which FE we currently connected
info.add(fe.getHost().equals(selfNode) ? "Yes" : "No");
info.add(fe.getIp().equals(selfNode) ? "Yes" : "No");
infos.add(info);
}
}
private static boolean isHelperNode(List<Pair<String, Integer>> helperNodes, Frontend fe) {
return helperNodes.stream().anyMatch(p -> p.first.equals(fe.getHost()) && p.second == fe.getEditLogPort());
private static boolean isHelperNode(List<HostInfo> helperNodes, Frontend fe) {
return helperNodes.stream().anyMatch(p -> p.getIp().equals(fe.getIp()) && p.getPort() == fe.getEditLogPort());
}
private static boolean isJoin(List<Pair<String, Integer>> allFeHosts, Frontend fe) {
for (Pair<String, Integer> pair : allFeHosts) {
if (fe.getHost().equals(pair.first) && fe.getEditLogPort() == pair.second) {
private static boolean isJoin(List<InetSocketAddress> allFeHosts, Frontend fe) {
for (InetSocketAddress addr : allFeHosts) {
if (fe.getEditLogPort() != addr.getPort()) {
continue;
}
if (!Strings.isNullOrEmpty(addr.getHostName())) {
if (addr.getHostName().equals(fe.getHostName())
|| addr.getHostName().equals(fe.getIp())) {
return true;
}
}
// if hostname of InetSocketAddress is ip, addr.getHostName() may be not equal to fe.getIp()
// so we need to compare fe.getIp() with address.getHostAddress()
InetAddress address = addr.getAddress();
if (null == address) {
LOG.warn("Failed to get InetAddress {}", addr);
continue;
}
if (fe.getIp().equals(address.getHostAddress())) {
return true;
}
}

2
fe/fe-core/src/main/java/org/apache/doris/common/telemetry/Telemetry.java
View File

@ -70,7 +70,7 @@ public class Telemetry {
throw new Exception("unknown value " + Config.trace_exporter + " of trace_exporter in fe.conf");
}
String serviceName = "FRONTEND:" + Env.getCurrentEnv().getSelfNode().first;
String serviceName = "FRONTEND:" + Env.getCurrentEnv().getSelfNode().getIp();
Resource serviceNameResource = Resource.create(
Attributes.of(AttributeKey.stringKey("service.name"), serviceName));
// Send a batch of spans if ScheduleDelay time or MaxExportBatchSize is reached

50
fe/fe-core/src/main/java/org/apache/doris/deploy/DeployManager.java
View File

@ -19,6 +19,7 @@ package org.apache.doris.deploy;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.FsBroker;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.Pair;
@ -28,6 +29,7 @@ import org.apache.doris.ha.FrontendNodeType;
import org.apache.doris.system.Backend;
import org.apache.doris.system.Frontend;
import org.apache.doris.system.SystemInfoService;
import org.apache.doris.system.SystemInfoService.HostInfo;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
@ -134,7 +136,8 @@ public class DeployManager extends MasterDaemon {
// So we use this map to count the continuous detected down times, if the continuous down time is more
// then MAX_MISSING_TIME, we considered this node as down permanently.
protected Map<String, Integer> counterMap = Maps.newHashMap();
protected static final Integer MAX_MISSING_TIME = 5;
// k8s pod delete and will recreate, so we need to wait for a while，otherwise we will drop node by mistake
protected static final Integer MAX_MISSING_TIME = 60;
public DeployManager(Env env, long intervalMs) {
super("deployManager", intervalMs);
@ -236,28 +239,20 @@ public class DeployManager extends MasterDaemon {
throw new NotImplementedException();
}
public List<Pair<String, Integer>> getHelperNodes() {
public List<HostInfo> getHelperNodes() {
String existFeHosts = System.getenv(ENV_FE_EXIST_ENDPOINT);
if (!Strings.isNullOrEmpty(existFeHosts)) {
// Some Frontends already exist in service group.
// We consider them as helper node
List<Pair<String, Integer>> helperNodes = Lists.newArrayList();
List<HostInfo> helperNodes = Lists.newArrayList();
String[] splittedHosts = existFeHosts.split(",");
for (String host : splittedHosts) {
String[] splittedHostPort = host.split(":");
if (splittedHostPort.length != 2) {
LOG.error("Invalid exist fe hosts: {}. will exit", existFeHosts);
System.exit(-1);
}
Integer port = -1;
try {
port = Integer.valueOf(splittedHostPort[1]);
} catch (NumberFormatException e) {
helperNodes.add(SystemInfoService.getIpHostAndPort(host, true));
} catch (AnalysisException e) {
LOG.error("Invalid exist fe hosts: {}. will exit", existFeHosts);
System.exit(-1);
}
helperNodes.add(Pair.of(splittedHostPort[0], port));
}
return helperNodes;
@ -327,8 +322,8 @@ public class DeployManager extends MasterDaemon {
LOG.info("sorted fe host list: {}", feHostInfos);
// 4. return the first one as helper
return Lists.newArrayList(Pair.of(feHostInfos.get(0).getIp(), feHostInfos.get(0).getPort()));
return Lists.newArrayList(new HostInfo(feHostInfos.get(0).getIp(), feHostInfos.get(0).getHostName(),
feHostInfos.get(0).getPort()));
}
@Override
@ -358,9 +353,8 @@ public class DeployManager extends MasterDaemon {
}
// 1.1 Check if self is in electable fe service group
// TODO(zd): 2023/2/17 Need to modify here when fe support FQDN (hostname will set to real hostname)
SystemInfoService.HostInfo selfHostInfo = getFromHostInfos(remoteElectableFeHosts,
new SystemInfoService.HostInfo(env.getMasterIp(), null, Config.edit_log_port));
new SystemInfoService.HostInfo(env.getMasterIp(), env.getMasterHostName(), Config.edit_log_port));
if (selfHostInfo == null) {
// The running of this deploy manager means this node is considered self as Master.
// If it self does not exist in electable fe service group, it should shut it self down.
@ -586,10 +580,10 @@ public class DeployManager extends MasterDaemon {
try {
switch (nodeType) {
case ELECTABLE:
env.dropFrontend(FrontendNodeType.FOLLOWER, localIp, localPort);
env.dropFrontend(FrontendNodeType.FOLLOWER, localIp, localHostName, localPort);
break;
case OBSERVER:
env.dropFrontend(FrontendNodeType.OBSERVER, localIp, localPort);
env.dropFrontend(FrontendNodeType.OBSERVER, localIp, localHostName, localPort);
break;
case BACKEND:
case BACKEND_CN:
@ -621,10 +615,10 @@ public class DeployManager extends MasterDaemon {
try {
switch (nodeType) {
case ELECTABLE:
env.addFrontend(FrontendNodeType.FOLLOWER, remoteIp, remotePort);
env.addFrontend(FrontendNodeType.FOLLOWER, remoteIp, remoteHostName, remotePort);
break;
case OBSERVER:
env.addFrontend(FrontendNodeType.OBSERVER, remoteIp, remotePort);
env.addFrontend(FrontendNodeType.OBSERVER, remoteIp, remoteHostName, remotePort);
break;
case BACKEND:
case BACKEND_CN:
@ -691,19 +685,23 @@ public class DeployManager extends MasterDaemon {
return hostPortPair;
}
// TODO: Need to modify here when fe support FQDN (hostname will set to real hostname)
private SystemInfoService.HostInfo convertToHostInfo(Frontend frontend) {
return new SystemInfoService.HostInfo(frontend.getHost(), null, frontend.getEditLogPort());
return new SystemInfoService.HostInfo(frontend.getIp(), frontend.getHostName(), frontend.getEditLogPort());
}
private SystemInfoService.HostInfo convertToHostInfo(Backend backend) {
return new SystemInfoService.HostInfo(backend.getIp(), backend.getHostName(), backend.getHeartbeatPort());
}
// TODO: Need to modify here when fe support FQDN(will check hostname?)
private boolean isSelf(SystemInfoService.HostInfo hostInfo) {
if (env.getMasterIp().equals(hostInfo.getIp()) && Config.edit_log_port == hostInfo.getPort()) {
return true;
if (Config.edit_log_port == hostInfo.getPort()) {
// master host name may not same as local host name, so we should compare ip here
if (env.getMasterHostName() != null && env.getMasterHostName().equals(hostInfo.getHostName())) {
return true;
}
if (env.getMasterIp().equals(hostInfo.getIp())) {
return true;
}
}
return false;
}

17
fe/fe-core/src/main/java/org/apache/doris/ha/BDBHA.java
View File

@ -221,6 +221,23 @@ public class BDBHA implements HAProtocol {
return true;
}
public boolean updateNodeAddress(String nodeName, String newHostName, int port) {
ReplicationGroupAdmin replicationGroupAdmin = environment.getReplicationGroupAdmin();
if (replicationGroupAdmin == null) {
return false;
}
try {
replicationGroupAdmin.updateAddress(nodeName, newHostName, port);
} catch (MemberNotFoundException e) {
LOG.error("the updating electable node is not found {}", nodeName, e);
return false;
} catch (MasterStateException e) {
LOG.error("the updating electable node is master {}", nodeName, e);
return false;
}
return true;
}
// When new Follower FE is added to the cluster, it should also be added to the
// helper sockets in
// ReplicationGroupAdmin, in order to fix the following case:

40
fe/fe-core/src/main/java/org/apache/doris/ha/MasterInfo.java
View File

@ -17,8 +17,13 @@
package org.apache.doris.ha;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.FeMetaVersion;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
import org.apache.doris.persist.gson.GsonUtils;
import com.google.gson.annotations.SerializedName;
import java.io.DataInput;
import java.io.DataOutput;
@ -26,18 +31,25 @@ import java.io.IOException;
public class MasterInfo implements Writable {
@SerializedName("ip")
private String ip;
@SerializedName("hostName")
private String hostName;
@SerializedName("httpPort")
private int httpPort;
@SerializedName("rpcPort")
private int rpcPort;
public MasterInfo() {
this.ip = "";
this.hostName = "";
this.httpPort = 0;
this.rpcPort = 0;
}
public MasterInfo(String ip, int httpPort, int rpcPort) {
public MasterInfo(String ip, String hostName, int httpPort, int rpcPort) {
this.ip = ip;
this.hostName = hostName;
this.httpPort = httpPort;
this.rpcPort = rpcPort;
}
@ -50,6 +62,14 @@ public class MasterInfo implements Writable {
this.ip = ip;
}
public String getHostName() {
return hostName;
}
public void setHostName(String hostName) {
this.hostName = hostName;
}
public int getHttpPort() {
return this.httpPort;
}
@ -68,15 +88,25 @@ public class MasterInfo implements Writable {
@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, ip);
out.writeInt(httpPort);
out.writeInt(rpcPort);
String json = GsonUtils.GSON.toJson(this);
Text.writeString(out, json);
}
public void readFields(DataInput in) throws IOException {
@Deprecated
private void readFields(DataInput in) throws IOException {
ip = Text.readString(in);
httpPort = in.readInt();
rpcPort = in.readInt();
}
public static MasterInfo read(DataInput in) throws IOException {
if (Env.getCurrentEnvJournalVersion() < FeMetaVersion.VERSION_118) {
MasterInfo masterInfo = new MasterInfo();
masterInfo.readFields(in);
return masterInfo;
}
String json = Text.readString(in);
return GsonUtils.GSON.fromJson(json, MasterInfo.class);
}
}

10
fe/fe-core/src/main/java/org/apache/doris/httpv2/meta/MetaService.java
View File

@ -54,13 +54,14 @@ public class MetaService extends RestBaseController {
private static final String VERSION = "version";
private static final String HOST = "host";
private static final String HOSTNAME = "hostname";
private static final String PORT = "port";
private File imageDir = MetaHelper.getMasterImageDir();
private boolean isFromValidFe(HttpServletRequest request) {
String clientHost = request.getRemoteHost();
Frontend fe = Env.getCurrentEnv().getFeByHost(clientHost);
Frontend fe = Env.getCurrentEnv().getFeByIp(clientHost);
if (fe == null) {
LOG.warn("request is not from valid FE. client: {}", clientHost);
return false;
@ -184,12 +185,15 @@ public class MetaService extends RestBaseController {
@RequestMapping(path = "/role", method = RequestMethod.GET)
public Object role(HttpServletRequest request, HttpServletResponse response) throws DdlException {
checkFromValidFe(request);
// For upgrade compatibility, the host parameter name remains the same
// and the new hostname parameter is added.
// host = ip
String host = request.getParameter(HOST);
String hostname = request.getParameter(HOSTNAME);
String portString = request.getParameter(PORT);
if (!Strings.isNullOrEmpty(host) && !Strings.isNullOrEmpty(portString)) {
int port = Integer.parseInt(portString);
Frontend fe = Env.getCurrentEnv().checkFeExist(host, port);
Frontend fe = Env.getCurrentEnv().checkFeExist(host, hostname, port);
if (fe == null) {
response.setHeader("role", FrontendNodeType.UNKNOWN.name());
} else {

2
fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/ClusterAction.java
View File

@ -60,7 +60,7 @@ public class ClusterAction extends RestBaseController {
Map<String, List<String>> result = Maps.newHashMap();
List<String> frontends = Env.getCurrentEnv().getFrontends(null)
.stream().filter(Frontend::isAlive)
.map(Frontend::getHost)
.map(Frontend::getIp)
.collect(Collectors.toList());
result.put("mysql", frontends.stream().map(ip -> ip + ":" + Config.query_port).collect(Collectors.toList()));

2
fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/HttpUtils.java
View File

@ -49,7 +49,7 @@ public class HttpUtils {
static List<Pair<String, Integer>> getFeList() {
return Env.getCurrentEnv().getFrontends(null)
.stream().filter(Frontend::isAlive).map(fe -> Pair.of(fe.getHost(), Config.http_port))
.stream().filter(Frontend::isAlive).map(fe -> Pair.of(fe.getIp(), Config.http_port))
.collect(Collectors.toList());
}

17
fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/NodeAction.java
View File

@ -21,7 +21,6 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.ConfigBase;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MarkedCountDownLatch;
import org.apache.doris.common.Pair;
import org.apache.doris.common.ThreadPoolManager;
@ -230,7 +229,7 @@ public class NodeAction extends RestBaseController {
}
private static List<String> getFeList() {
return Env.getCurrentEnv().getFrontends(null).stream().map(fe -> fe.getHost() + ":" + Config.http_port)
return Env.getCurrentEnv().getFrontends(null).stream().map(fe -> fe.getIp() + ":" + Config.http_port)
.collect(Collectors.toList());
}
@ -483,7 +482,7 @@ public class NodeAction extends RestBaseController {
List<Map<String, String>> failedTotal = Lists.newArrayList();
List<NodeConfigs> nodeConfigList = parseSetConfigNodes(requestBody, failedTotal);
List<Pair<String, Integer>> aliveFe = Env.getCurrentEnv().getFrontends(null).stream().filter(Frontend::isAlive)
.map(fe -> Pair.of(fe.getHost(), Config.http_port)).collect(Collectors.toList());
.map(fe -> Pair.of(fe.getIp(), Config.http_port)).collect(Collectors.toList());
checkNodeIsAlive(nodeConfigList, aliveFe, failedTotal);
Map<String, String> header = Maps.newHashMap();
@ -646,9 +645,6 @@ public class NodeAction extends RestBaseController {
}
try {
String role = reqInfo.getRole();
String[] split = reqInfo.getHostPort().split(":");
String host = split[0];
int port = Integer.parseInt(split[1]);
Env currentEnv = Env.getCurrentEnv();
FrontendNodeType frontendNodeType;
if (FrontendNodeType.FOLLOWER.name().equals(role)) {
@ -656,13 +652,14 @@ public class NodeAction extends RestBaseController {
} else {
frontendNodeType = FrontendNodeType.OBSERVER;
}
HostInfo info = SystemInfoService.getIpHostAndPort(reqInfo.getHostPort(), true);
if ("ADD".equals(action)) {
currentEnv.addFrontend(frontendNodeType, host, port);
currentEnv.addFrontend(frontendNodeType, info.getIp(), info.getHostName(), info.getPort());
} else if ("DROP".equals(action)) {
currentEnv.dropFrontend(frontendNodeType, host, port);
currentEnv.dropFrontend(frontendNodeType, info.getIp(), info.getHostName(), info.getPort());
}
} catch (DdlException ddlException) {
return ResponseEntityBuilder.okWithCommonError(ddlException.getMessage());
} catch (UserException userException) {
return ResponseEntityBuilder.okWithCommonError(userException.getMessage());
}
return ResponseEntityBuilder.ok();
}

2
fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/manager/QueryProfileAction.java
View File

@ -175,7 +175,7 @@ public class QueryProfileAction extends RestBaseController {
// add node information
for (List<String> query : queries) {
query.add(1, Env.getCurrentEnv().getSelfNode().first + ":" + Config.http_port);
query.add(1, Env.getCurrentEnv().getSelfNode().getIp() + ":" + Config.http_port);
}
if (!Strings.isNullOrEmpty(search)) {

7
fe/fe-core/src/main/java/org/apache/doris/journal/JournalEntity.java
View File

@ -337,9 +337,9 @@ public class JournalEntity implements Writable {
}
case OperationType.OP_ADD_FRONTEND:
case OperationType.OP_ADD_FIRST_FRONTEND:
case OperationType.OP_MODIFY_FRONTEND:
case OperationType.OP_REMOVE_FRONTEND: {
data = new Frontend();
((Frontend) data).readFields(in);
data = Frontend.read(in);
isRead = true;
break;
}
@ -375,8 +375,7 @@ public class JournalEntity implements Writable {
break;
}
case OperationType.OP_MASTER_INFO_CHANGE: {
data = new MasterInfo();
((MasterInfo) data).readFields(in);
data = MasterInfo.read(in);
isRead = true;
break;
}

1
fe/fe-core/src/main/java/org/apache/doris/journal/bdbje/BDBEnvironment.java
View File

@ -183,7 +183,6 @@ public class BDBEnvironment {
// start state change listener
StateChangeListener listener = new BDBStateChangeListener();
replicatedEnvironment.setStateChangeListener(listener);
// open epochDB. the first parameter null means auto-commit
epochDB = replicatedEnvironment.openDatabase(null, "epochDB", dbConfig);
break;

26
fe/fe-core/src/main/java/org/apache/doris/journal/bdbje/BDBJEJournal.java
View File

@ -18,7 +18,7 @@
package org.apache.doris.journal.bdbje;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.Pair;
import org.apache.doris.common.Config;
import org.apache.doris.common.io.DataOutputBuffer;
import org.apache.doris.common.io.Writable;
import org.apache.doris.common.util.Util;
@ -27,6 +27,7 @@ import org.apache.doris.journal.JournalCursor;
import org.apache.doris.journal.JournalEntity;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.persist.OperationType;
import org.apache.doris.system.SystemInfoService.HostInfo;
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.je.Database;
@ -80,9 +81,17 @@ public class BDBJEJournal implements Journal { // CHECKSTYLE IGNORE THIS LINE: B
*/
private void initBDBEnv(String nodeName) {
environmentPath = Env.getServingEnv().getBdbDir();
Pair<String, Integer> selfNode = Env.getServingEnv().getSelfNode();
HostInfo selfNode = Env.getServingEnv().getSelfNode();
selfNodeName = nodeName;
selfNodeHostPort = selfNode.first + ":" + selfNode.second;
if (Config.enable_fqdn_mode) {
// We use the hostname as the address of the bdbje node,
// so that we do not need to update bdbje when the IP changes.
// WARNING:However, it is necessary to ensure that the hostname of the node
// can be resolved and accessed by other nodes.
selfNodeHostPort = selfNode.getHostName() + ":" + selfNode.getPort();
} else {
selfNodeHostPort = selfNode.getIp() + ":" + selfNode.getPort();
}
}
/*
@ -299,8 +308,11 @@ public class BDBJEJournal implements Journal { // CHECKSTYLE IGNORE THIS LINE: B
if (bdbEnvironment == null) {
File dbEnv = new File(environmentPath);
bdbEnvironment = new BDBEnvironment();
Pair<String, Integer> helperNode = Env.getServingEnv().getHelperNode();
String helperHostPort = helperNode.first + ":" + helperNode.second;
HostInfo helperNode = Env.getServingEnv().getHelperNode();
String helperHostPort = helperNode.getIp() + ":" + helperNode.getPort();
if (Config.enable_fqdn_mode) {
helperHostPort = helperNode.getHostName() + ":" + helperNode.getPort();
}
try {
bdbEnvironment.setup(dbEnv, selfNodeName, selfNodeHostPort, helperHostPort,
Env.getServingEnv().isElectable());
@ -358,14 +370,14 @@ public class BDBJEJournal implements Journal { // CHECKSTYLE IGNORE THIS LINE: B
// the files
// ATTN: here we use `getServingEnv()`, because only serving catalog has
// helper nodes.
Pair<String, Integer> helperNode = Env.getServingEnv().getHelperNode();
HostInfo helperNode = Env.getServingEnv().getHelperNode();
NetworkRestore restore = new NetworkRestore();
NetworkRestoreConfig config = new NetworkRestoreConfig();
config.setRetainLogFiles(false);
restore.execute(insufficientLogEx, config);
bdbEnvironment.close();
bdbEnvironment.setup(new File(environmentPath), selfNodeName, selfNodeHostPort,
helperNode.first + ":" + helperNode.second, Env.getServingEnv().isElectable());
helperNode.getIp() + ":" + helperNode.getPort(), Env.getServingEnv().isElectable());
}
@Override

4
fe/fe-core/src/main/java/org/apache/doris/master/Checkpoint.java
View File

@ -185,7 +185,7 @@ public class Checkpoint extends MasterDaemon {
if (!allFrontends.isEmpty()) {
otherNodesCount = allFrontends.size() - 1; // skip master itself
for (Frontend fe : allFrontends) {
String host = fe.getHost();
String host = fe.getIp();
if (host.equals(Env.getServingEnv().getMasterIp())) {
// skip master itself
continue;
@ -227,7 +227,7 @@ public class Checkpoint extends MasterDaemon {
long deleteVersion = storage.getLatestValidatedImageSeq();
if (successPushed > 0) {
for (Frontend fe : allFrontends) {
String host = fe.getHost();
String host = fe.getIp();
if (host.equals(Env.getServingEnv().getMasterIp())) {
// skip master itself
continue;

9
fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
View File

@ -402,6 +402,11 @@ public class EditLog {
env.replayAddFrontend(fe);
break;
}
case OperationType.OP_MODIFY_FRONTEND: {
Frontend fe = (Frontend) journal.getData();
env.replayModifyFrontend(fe);
break;
}
case OperationType.OP_REMOVE_FRONTEND: {
Frontend fe = (Frontend) journal.getData();
env.replayDropFrontend(fe);
@ -1236,6 +1241,10 @@ public class EditLog {
logEdit(OperationType.OP_ADD_FIRST_FRONTEND, fe);
}
public void logModifyFrontend(Frontend fe) {
logEdit(OperationType.OP_MODIFY_FRONTEND, fe);
}
public void logRemoveFrontend(Frontend fe) {
logEdit(OperationType.OP_REMOVE_FRONTEND, fe);
}

2
fe/fe-core/src/main/java/org/apache/doris/persist/OperationType.java
View File

@ -150,6 +150,8 @@ public class OperationType {
public static final short OP_DROP_REPOSITORY = 90;
public static final short OP_MODIFY_BACKEND = 91;
public static final short OP_MODIFY_FRONTEND = 92;
//colocate table
public static final short OP_COLOCATE_ADD_TABLE = 94;
public static final short OP_COLOCATE_REMOVE_TABLE = 95;

2
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
View File

@ -497,7 +497,7 @@ public class StmtExecutor implements ProfileWriter {
// If goes here, which means we can't find a valid Master FE(some error happens).
// To avoid endless forward, throw exception here.
throw new UserException("The statement has been forwarded to master FE("
+ Env.getCurrentEnv().getSelfNode().first + ") and failed to execute"
+ Env.getCurrentEnv().getSelfNode().getIp() + ") and failed to execute"
+ " because Master FE is not ready. You may need to check FE's status");
}
forwardToMaster();

17
fe/fe-core/src/main/java/org/apache/doris/service/FrontendOptions.java
View File

@ -83,6 +83,8 @@ public class FrontendOptions {
if (localAddr == null) {
localAddr = loopBack;
}
checkHostName();
LOG.info("local address: {}.", localAddr);
}
@ -94,10 +96,23 @@ public class FrontendOptions {
return InetAddresses.toAddrString(localAddr);
}
public static String getHostname() {
public static String getHostName() {
return localAddr.getHostName();
}
private static void checkHostName() throws UnknownHostException {
if (Config.enable_fqdn_mode) {
if (getHostName().equals(getLocalHostAddress())) {
LOG.error("Can't get hostname in FQDN mode. Please check your network configuration."
+ " got hostname: {}, ip: {}",
getHostName(), getLocalHostAddress());
throw new UnknownHostException("Can't get hostname in FQDN mode."
+ " Please check your network configuration."
+ " got hostname: " + getHostName() + ", ip: " + getLocalHostAddress());
}
}
}
private static void analyzePriorityCidrs() {
String priorCidrs = Config.priority_networks;
if (Strings.isNullOrEmpty(priorCidrs)) {

2
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
View File

@ -786,7 +786,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
public TMasterOpResult forward(TMasterOpRequest params) throws TException {
TNetworkAddress clientAddr = getClientAddr();
if (clientAddr != null) {
Frontend fe = Env.getCurrentEnv().getFeByHost(clientAddr.getHostname());
Frontend fe = Env.getCurrentEnv().getFeByIp(clientAddr.getHostname());
if (fe == null) {
LOG.warn("reject request from invalid host. client: {}", clientAddr);
throw new TException("request from invalid host was rejected.");

30
fe/fe-core/src/main/java/org/apache/doris/system/FQDNManager.java
View File

@ -19,10 +19,12 @@ package org.apache.doris.system;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.ClientPool;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.thrift.TNetworkAddress;
import com.google.common.base.Strings;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -40,10 +42,36 @@ public class FQDNManager extends MasterDaemon {
}
/**
* At each round: check if ip of be has already been changed
* At each round: check if ip of be or fe has already been changed
*/
@Override
protected void runAfterCatalogReady() {
updateBeIp();
updateFeIp();
}
private void updateFeIp() {
for (Frontend fe : Env.getCurrentEnv().getFrontends(null /* all */)) {
if (!Strings.isNullOrEmpty(fe.getHostName())) {
try {
InetAddress inetAddress = InetAddress.getByName(fe.getHostName());
if (!fe.getIp().equalsIgnoreCase(inetAddress.getHostAddress())) {
String oldIp = fe.getIp();
String newIp = inetAddress.getHostAddress();
Env.getCurrentEnv().modifyFrontendIp(fe.getNodeName(), newIp);
LOG.warn("ip for {} of fe has been changed from {} to {}",
fe.getHostName(), oldIp, fe.getIp());
}
} catch (UnknownHostException e) {
LOG.warn("unknown host name for fe, {}", fe.getHostName(), e);
} catch (DdlException e) {
LOG.warn("fail to update ip for fe, {}", fe.getHostName(), e);
}
}
}
}
private void updateBeIp() {
for (Backend be : nodeMgr.getIdToBackend().values()) {
if (be.getHostName() != null) {
try {

63
fe/fe-core/src/main/java/org/apache/doris/system/Frontend.java
View File

@ -19,20 +19,32 @@ package org.apache.doris.system;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeMetaVersion;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
import org.apache.doris.ha.BDBHA;
import org.apache.doris.ha.FrontendNodeType;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.system.HeartbeatResponse.HbStatus;
import com.google.gson.annotations.SerializedName;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class Frontend implements Writable {
@SerializedName("role")
private FrontendNodeType role;
// nodeName = ip:port_timestamp
@SerializedName("nodeName")
private String nodeName;
private String host;
@SerializedName("ip")
private volatile String ip;
// used for getIpByHostname
@SerializedName("hostName")
private String hostName;
@SerializedName("editLogPort")
private int editLogPort;
private String version;
@ -47,10 +59,15 @@ public class Frontend implements Writable {
public Frontend() {}
public Frontend(FrontendNodeType role, String nodeName, String host, int editLogPort) {
public Frontend(FrontendNodeType role, String nodeName, String ip, int editLogPort) {
this(role, nodeName, ip, "", editLogPort);
}
public Frontend(FrontendNodeType role, String nodeName, String ip, String hostName, int editLogPort) {
this.role = role;
this.nodeName = nodeName;
this.host = host;
this.ip = ip;
this.hostName = hostName;
this.editLogPort = editLogPort;
}
@ -58,14 +75,22 @@ public class Frontend implements Writable {
return this.role;
}
public String getHost() {
return this.host;
public String getIp() {
return this.ip;
}
public String getVersion() {
return version;
}
public String getHostName() {
return hostName;
}
public void setHostName(String hostName) {
this.hostName = hostName;
}
public String getNodeName() {
return nodeName;
}
@ -131,35 +156,43 @@ public class Frontend implements Writable {
@Override
public void write(DataOutput out) throws IOException {
Text.writeString(out, role.name());
Text.writeString(out, host);
out.writeInt(editLogPort);
Text.writeString(out, nodeName);
String json = GsonUtils.GSON.toJson(this);
Text.writeString(out, json);
}
public void readFields(DataInput in) throws IOException {
@Deprecated
private void readFields(DataInput in) throws IOException {
role = FrontendNodeType.valueOf(Text.readString(in));
if (role == FrontendNodeType.REPLICA) {
// this is for compatibility.
// we changed REPLICA to FOLLOWER
role = FrontendNodeType.FOLLOWER;
}
host = Text.readString(in);
ip = Text.readString(in);
editLogPort = in.readInt();
nodeName = Text.readString(in);
}
public static Frontend read(DataInput in) throws IOException {
Frontend frontend = new Frontend();
frontend.readFields(in);
return frontend;
if (Env.getCurrentEnvJournalVersion() < FeMetaVersion.VERSION_118) {
Frontend frontend = new Frontend();
frontend.readFields(in);
return frontend;
}
String json = Text.readString(in);
return GsonUtils.GSON.fromJson(json, Frontend.class);
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("name: ").append(nodeName).append(", role: ").append(role.name());
sb.append(", ").append(host).append(":").append(editLogPort);
sb.append(", hostname: ").append(hostName);
sb.append(", ").append(ip).append(":").append(editLogPort);
return sb.toString();
}
public void setIp(String ip) {
this.ip = ip;
}
}

7
fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java
View File

@ -29,6 +29,7 @@ import org.apache.doris.persist.HbPackage;
import org.apache.doris.resource.Tag;
import org.apache.doris.service.FrontendOptions;
import org.apache.doris.system.HeartbeatResponse.HbStatus;
import org.apache.doris.system.SystemInfoService.HostInfo;
import org.apache.doris.thrift.FrontendService;
import org.apache.doris.thrift.HeartbeatService;
import org.apache.doris.thrift.TBackendInfo;
@ -293,7 +294,9 @@ public class HeartbeatMgr extends MasterDaemon {
@Override
public HeartbeatResponse call() {
if (fe.getHost().equals(Env.getCurrentEnv().getSelfNode().first)) {
HostInfo selfNode = Env.getCurrentEnv().getSelfNode();
if (fe.getIp().equals(selfNode.getIp())
|| (!Strings.isNullOrEmpty(fe.getHostName()) && fe.getHostName().equals(selfNode.getHostName()))) {
// heartbeat to self
if (Env.getCurrentEnv().isReady()) {
return new FrontendHbResponse(fe.getNodeName(), Config.query_port, Config.rpc_port,
@ -309,7 +312,7 @@ public class HeartbeatMgr extends MasterDaemon {
private HeartbeatResponse getHeartbeatResponse() {
FrontendService.Client client = null;
TNetworkAddress addr = new TNetworkAddress(fe.getHost(), Config.rpc_port);
TNetworkAddress addr = new TNetworkAddress(fe.getIp(), Config.rpc_port);
boolean ok = false;
try {
client = ClientPool.frontendHeartbeatPool.borrowObject(addr);

40
fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
View File

@ -61,6 +61,7 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
@ -127,6 +128,33 @@ public class SystemInfoService {
return res;
}
public boolean isSame(HostInfo other) {
if (other.getPort() != port) {
return false;
}
if (hostName != null && hostName.equals(other.getHostName())) {
return true;
}
if (ip != null && ip.equals(other.getIp())) {
return true;
}
return false;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
HostInfo that = (HostInfo) o;
return Objects.equals(ip, that.getIp())
&& Objects.equals(hostName, that.getHostName())
&& Objects.equals(port, that.getPort());
}
@Override
public String toString() {
return "HostInfo{"
@ -168,8 +196,8 @@ public class SystemInfoService {
if (!Config.enable_fqdn_mode) {
hostInfo.setHostName(null);
}
if (Config.enable_fqdn_mode && hostInfo.getHostName() == null) {
throw new DdlException("backend's hostName should not be null while enable_fqdn_mode is true");
if (Config.enable_fqdn_mode && StringUtils.isEmpty(hostInfo.getHostName())) {
throw new DdlException("backend's hostName should not be empty while enable_fqdn_mode is true");
}
// check is already exist
if (getBackendWithHeartbeatPort(hostInfo.getIp(), hostInfo.getHostName(), hostInfo.getPort()) != null) {
@ -355,20 +383,20 @@ public class SystemInfoService {
return null;
}
public Backend getBackendWithBePort(String host, int bePort) {
public Backend getBackendWithBePort(String ip, int bePort) {
ImmutableMap<Long, Backend> idToBackend = idToBackendRef;
for (Backend backend : idToBackend.values()) {
if (backend.getIp().equals(host) && backend.getBePort() == bePort) {
if (backend.getIp().equals(ip) && backend.getBePort() == bePort) {
return backend;
}
}
return null;
}
public Backend getBackendWithHttpPort(String host, int httpPort) {
public Backend getBackendWithHttpPort(String ip, int httpPort) {
ImmutableMap<Long, Backend> idToBackend = idToBackendRef;
for (Backend backend : idToBackend.values()) {
if (backend.getIp().equals(host) && backend.getHttpPort() == httpPort) {
if (backend.getIp().equals(ip) && backend.getHttpPort() == httpPort) {
return backend;
}
}

69
fe/fe-core/src/test/java/org/apache/doris/system/FQDNManagerTest.java
View File

@ -19,6 +19,8 @@ package org.apache.doris.system;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.ha.BDBHA;
import org.apache.doris.ha.FrontendNodeType;
import mockit.Expectations;
import mockit.Mock;
@ -30,6 +32,8 @@ import org.junit.Test;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
public class FQDNManagerTest {
@Mocked
@ -42,6 +46,11 @@ public class FQDNManagerTest {
private SystemInfoService systemInfoService;
List<Frontend> frontends = new ArrayList<>();
@Mocked
private BDBHA bdbha;
@Before
public void setUp() throws UnknownHostException {
new MockUp<InetAddress>(InetAddress.class) {
@ -56,8 +65,41 @@ public class FQDNManagerTest {
public Env getServingEnv() {
return env;
}
@Mock
public Env getCurrentEnv() {
return env;
}
@Mock
public boolean tryLock() {
return true;
}
@Mock
public void modifyFrontendIp(String nodeName, String ip) {
for (Frontend fe : frontends) {
if (fe.getNodeName().equals(nodeName)) {
fe.setIp(ip);
}
}
}
};
new MockUp<BDBHA>(BDBHA.class) {
@Mock
public boolean updateNodeAddress(String nodeName, String ip, int port) {
return true;
}
};
Config.enable_fqdn_mode = true;
systemInfoService = new SystemInfoService();
systemInfoService.addBackend(new Backend(1, "193.88.67.98", "doris.test.domain", 9090));
frontends.add(new Frontend(FrontendNodeType.FOLLOWER, "doris.test.domain_9010_1675168383846",
"193.88.67.90", "doris.test.domain", 9010));
fdqnManager = new FQDNManager(systemInfoService);
new Expectations() {
{
env.isReady();
@ -67,13 +109,20 @@ public class FQDNManagerTest {
inetAddress.getHostAddress();
minTimes = 0;
result = "193.88.67.99";
env.getFrontends(null);
minTimes = 0;
result = frontends;
env.getFeByName("doris.test.domain_9010_1675168383846");
minTimes = 0;
result = frontends.get(0);
env.getHaProtocol();
minTimes = 0;
result = bdbha;
}
};
Config.enable_fqdn_mode = true;
systemInfoService = new SystemInfoService();
systemInfoService.addBackend(new Backend(1, "193.88.67.98", "doris.test.domain", 9090));
fdqnManager = new FQDNManager(systemInfoService);
}
@Test
@ -84,4 +133,14 @@ public class FQDNManagerTest {
Assert.assertEquals("193.88.67.99", systemInfoService.getBackend(1).getIp());
fdqnManager.exit();
}
@Test
public void testFrontendChanged() throws InterruptedException {
// frontend ip changed
Assert.assertEquals("193.88.67.90", env.getFrontends(null).get(0).getIp());
fdqnManager.start();
Thread.sleep(1000);
Assert.assertEquals("193.88.67.99", env.getFrontends(null).get(0).getIp());
fdqnManager.exit();
}
}

4
fe/fe-core/src/test/java/org/apache/doris/system/HeartbeatMgrTest.java
View File

@ -20,11 +20,11 @@ package org.apache.doris.system;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.FsBroker;
import org.apache.doris.common.GenericPool;
import org.apache.doris.common.Pair;
import org.apache.doris.ha.FrontendNodeType;
import org.apache.doris.system.HeartbeatMgr.BrokerHeartbeatHandler;
import org.apache.doris.system.HeartbeatMgr.FrontendHeartbeatHandler;
import org.apache.doris.system.HeartbeatResponse.HbStatus;
import org.apache.doris.system.SystemInfoService.HostInfo;
import org.apache.doris.thrift.FrontendService;
import org.apache.doris.thrift.TBrokerOperationStatus;
import org.apache.doris.thrift.TBrokerOperationStatusCode;
@ -55,7 +55,7 @@ public class HeartbeatMgrTest {
{
env.getSelfNode();
minTimes = 0;
result = Pair.of("192.168.1.3", 9010); // not self
result = new HostInfo("192.168.1.3", null, 9010); // not self
env.isReady();
minTimes = 0;