This is an automated email from the ASF dual-hosted git repository.
neuyilan pushed a commit to branch change_rpc_port
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/change_rpc_port by this push:
new 9c90546 change seed_nodes only contains the interal_ip and
meta_port&exchange the nodes info when first start start during the meta group
established
9c90546 is described below
commit 9c9054666ec2647fcb78fd0a6b5daffbc4f9a7cf
Author: HouliangQi <[email protected]>
AuthorDate: Thu Feb 25 17:47:09 2021 +0800
change seed_nodes only contains the interal_ip and meta_port&exchange the
nodes info when first start start during the meta group established
---
.../resources/conf/iotdb-cluster.properties | 6 +-
.../java/org/apache/iotdb/cluster/ClusterMain.java | 7 ++-
.../apache/iotdb/cluster/config/ClusterConfig.java | 5 +-
.../iotdb/cluster/config/ClusterDescriptor.java | 28 +++++++---
.../iotdb/cluster/coordinator/Coordinator.java | 2 +-
.../cluster/partition/slot/SlotPartitionTable.java | 3 -
.../apache/iotdb/cluster/server/RaftServer.java | 3 +
.../cluster/server/heartbeat/HeartbeatServer.java | 2 +
.../cluster/server/member/DataGroupMember.java | 3 +-
.../cluster/server/member/MetaGroupMember.java | 65 ++++++++++++++++++----
.../cluster/server/service/BaseAsyncService.java | 3 +-
.../apache/iotdb/cluster/utils/ClusterNode.java | 13 +++--
.../apache/iotdb/cluster/utils/ClusterUtils.java | 20 +++----
.../cluster/common/TestPartitionedLogManager.java | 8 +--
.../org/apache/iotdb/cluster/common/TestUtils.java | 2 +
.../apache/iotdb/cluster/log/HardStateTest.java | 5 +-
.../cluster/log/applier/MetaLogApplierTest.java | 4 +-
.../cluster/log/logtypes/SerializeLogTest.java | 4 +-
.../cluster/partition/SlotPartitionTableTest.java | 61 ++++++++++++++------
.../handlers/caller/HeartbeatHandlerTest.java | 8 +--
.../cluster/server/member/DataGroupMemberTest.java | 10 +---
.../cluster/server/member/MetaGroupMemberTest.java | 5 +-
.../org/apache/iotdb/cluster/utils/Constants.java | 1 -
.../resources/node1conf/iotdb-cluster.properties | 2 +-
.../resources/node2conf/iotdb-cluster.properties | 2 +-
.../resources/node3conf/iotdb-cluster.properties | 2 +-
.../org/apache/iotdb/db/utils/SerializeUtils.java | 21 ++++---
.../apache/iotdb/db/utils/SerializeUtilsTest.java | 4 +-
thrift/src/main/thrift/cluster.thrift | 5 +-
29 files changed, 195 insertions(+), 109 deletions(-)
diff --git a/cluster/src/assembly/resources/conf/iotdb-cluster.properties
b/cluster/src/assembly/resources/conf/iotdb-cluster.properties
index a32daec..e754657 100644
--- a/cluster/src/assembly/resources/conf/iotdb-cluster.properties
+++ b/cluster/src/assembly/resources/conf/iotdb-cluster.properties
@@ -34,7 +34,7 @@ internal_meta_port=9003
internal_data_port=40010
# whether open port for server module (for debug purpose)
-# if true, the ip will be rpc_port (in iotdb-engines.properties) + 1
+# if true, the single's server rpc_port will be changed to rpc_port (in
iotdb-engines.properties) + 1
open_server_rpc_port=false
# comma-separated {IP/DOMAIN}:meta_port:data_port:client_port pairs
@@ -47,7 +47,7 @@ open_server_rpc_port=false
# nodes that already in the cluster, unnecessary to be the nodes that were
used to build the
# initial cluster by start-node.sh(.bat). Several nodes will be picked
randomly to send the
# request, the number of nodes picked depends on the number of retries.
-seed_nodes=127.0.0.1:9003:40010:6667,127.0.0.1:9005:40012:6668,127.0.0.1:9007:40014:6669
+seed_nodes=127.0.0.1:9003,127.0.0.1:9005,127.0.0.1:9007
# whether to use thrift compressed protocol for internal communications. If
you want to change
# compression settings for external clients, please modify
'rpc_thrift_compression_enable' in
@@ -61,7 +61,7 @@ rpc_thrift_compression_enable=false
max_concurrent_client_num=10000
# number of replications for one partition
-default_replica_num=2
+default_replica_num=3
# cluster name to identify different clusters
# all node's cluster_name in one cluster are the same
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java
b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java
index 63f2764..4332d3c 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java
@@ -34,6 +34,7 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
+
import org.apache.thrift.TException;
import org.apache.thrift.async.TAsyncClientManager;
import org.apache.thrift.protocol.TBinaryProtocol.Factory;
@@ -76,7 +77,6 @@ public class ClusterMain {
// init server's configuration first, because the cluster configuration
may read settings from
// the server's configuration.
- IoTDBDescriptor.getInstance().getConfig().setEnableRPCService(false);
IoTDBDescriptor.getInstance().getConfig().setSyncEnable(false);
// auto create schema is took over by cluster module, so we disable it in
the server module.
IoTDBDescriptor.getInstance().getConfig().setAutoCreateSchemaEnabled(false);
@@ -180,10 +180,11 @@ public class ClusterMain {
// assert this node is in seed nodes list
Node localNode = new Node();
localNode
- .setIp(IoTDBDescriptor.getInstance().getConfig().getRpcAddress())
+ .setIp(config.getInternalIp())
.setMetaPort(config.getInternalMetaPort())
.setDataPort(config.getInternalDataPort())
- .setClientPort(config.getClusterRpcPort());
+ .setClientPort(config.getClusterRpcPort())
+
.setClientIp(IoTDBDescriptor.getInstance().getConfig().getRpcAddress());
if (!seedNodes.contains(localNode)) {
String message =
String.format(
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
index dd51eeb..8ba7552 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
@@ -35,13 +35,12 @@ public class ClusterConfig {
/** each one is a "<IP | domain name>:<meta port>:<data port>:<client
port></>" string tuple */
private List<String> seedNodeUrls =
- Arrays.asList(
- String.format("127.0.0.1:%d:%d:%d", internalMetaPort,
internalDataPort, clusterRpcPort));
+ Arrays.asList(String.format("%s:%d", internalIp, internalMetaPort));
@ClusterConsistent private boolean isRpcThriftCompressionEnabled = false;
private int maxConcurrentClientNum = 10000;
- @ClusterConsistent private int replicationNum = 2;
+ @ClusterConsistent private int replicationNum = 3;
@ClusterConsistent private String clusterName = "default";
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
index c501122..98ae042 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
@@ -19,11 +19,12 @@
package org.apache.iotdb.cluster.config;
-import com.google.common.net.InetAddresses;
import org.apache.iotdb.cluster.exception.BadSeedUrlFormatException;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.query.QueryProcessException;
+
+import com.google.common.net.InetAddresses;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,12 +43,21 @@ public class ClusterDescriptor {
private static final Logger logger =
LoggerFactory.getLogger(ClusterDescriptor.class);
private static final ClusterDescriptor INSTANCE = new ClusterDescriptor();
-
- private ClusterConfig config = new ClusterConfig();
+ private final ClusterConfig config = new ClusterConfig();
private ClusterDescriptor() {
// copy needed configurations from the server's config to the cluster.
config.setClusterRpcPort(IoTDBDescriptor.getInstance().getConfig().getRpcPort());
+ // if open the server rpc port, we will enable the rpc service and change
the server's rpc port
+ // to rpc_port + 1
+ if (config.isOpenServerRpcPort()) {
+ IoTDBDescriptor.getInstance().getConfig().setEnableRPCService(true);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setRpcPort(IoTDBDescriptor.getInstance().getConfig().getRpcPort() +
1);
+ } else {
+ IoTDBDescriptor.getInstance().getConfig().setEnableRPCService(false);
+ }
// then load settings from cluster's file.
// so, iotdb-cluster.properties can overwrite iotdb-properties.
loadProps();
@@ -82,7 +92,6 @@ public class ClusterDescriptor {
}
public void replaceHostnameWithIp() throws UnknownHostException,
BadSeedUrlFormatException {
-
boolean isInvalidClusterInternalIp =
InetAddresses.isInetAddress(config.getInternalIp());
if (!isInvalidClusterInternalIp) {
config.setInternalIp(hostnameToIP(config.getInternalIp()));
@@ -90,21 +99,21 @@ public class ClusterDescriptor {
List<String> newSeedUrls = new ArrayList<>();
for (String seedUrl : config.getSeedNodeUrls()) {
String[] splits = seedUrl.split(":");
- if (splits.length != 4) {
+ if (splits.length != 2) {
throw new BadSeedUrlFormatException(seedUrl);
}
String seedIP = splits[0];
boolean isInvalidSeedIp = InetAddresses.isInetAddress(seedIP);
if (!isInvalidSeedIp) {
String newSeedIP = hostnameToIP(seedIP);
- newSeedUrls.add(newSeedIP + ":" + splits[1] + ":" + splits[2] + ":" +
splits[3]);
+ newSeedUrls.add(newSeedIP + ":" + splits[1]);
} else {
newSeedUrls.add(seedUrl);
}
}
config.setSeedNodeUrls(newSeedUrls);
logger.debug(
- "after replace, the rpcIP={}, internalIP={} seedUrls={}",
+ "after replace, the rpcIP={}, internalIP={}, seedUrls={}",
IoTDBDescriptor.getInstance().getConfig().getRpcAddress(),
config.getInternalIp(),
config.getSeedNodeUrls());
@@ -209,6 +218,11 @@ public class ClusterDescriptor {
properties.getProperty(
"is_use_async_server",
String.valueOf(config.isUseAsyncServer()))));
+ config.setOpenServerRpcPort(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "open_server_rpc_port",
String.valueOf(config.isOpenServerRpcPort()))));
+
config.setUseAsyncApplier(
Boolean.parseBoolean(
properties.getProperty(
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
b/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
index 0ad14f4..d8b6e7a 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/coordinator/Coordinator.java
@@ -595,7 +595,7 @@ public class Coordinator {
}
if (!StatusUtils.TIME_OUT.equals(status)) {
if (!status.isSetRedirectNode()) {
- status.setRedirectNode(new EndPoint(node.getIp(),
node.getClientPort()));
+ status.setRedirectNode(new EndPoint(node.getClientIp(),
node.getClientPort()));
}
return status;
} else {
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotPartitionTable.java
b/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotPartitionTable.java
index 920681a..8c25a63 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotPartitionTable.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotPartitionTable.java
@@ -303,7 +303,6 @@ public class SlotPartitionTable implements PartitionTable {
@Override
public ByteBuffer serialize() {
-
ByteArrayOutputStream outputStream = new ByteArrayOutputStream(4096);
DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
@@ -326,7 +325,6 @@ public class SlotPartitionTable implements PartitionTable {
dataOutputStream.writeInt(integerNodeEntry.getValue().getNodeIdentifier());
}
}
-
dataOutputStream.writeLong(lastLogIndex);
} catch (IOException ignored) {
// not reachable
@@ -336,7 +334,6 @@ public class SlotPartitionTable implements PartitionTable {
@Override
public void deserialize(ByteBuffer buffer) {
-
logger.info("Initializing the partition table from buffer");
totalSlotNumbers = buffer.getInt();
int size = buffer.getInt();
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/RaftServer.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/RaftServer.java
index e36264f..f77f29a 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/server/RaftServer.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/RaftServer.java
@@ -83,10 +83,13 @@ public abstract class RaftServer implements
RaftService.AsyncIface, RaftService.
RaftServer() {
thisNode = new Node();
+ // set internal rpc ip and ports
thisNode.setIp(config.getInternalIp());
thisNode.setMetaPort(config.getInternalMetaPort());
thisNode.setDataPort(config.getInternalDataPort());
+ // set client rpc ip and ports
thisNode.setClientPort(config.getClusterRpcPort());
+
thisNode.setClientIp(IoTDBDescriptor.getInstance().getConfig().getRpcAddress());
}
RaftServer(Node thisNode) {
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/heartbeat/HeartbeatServer.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/heartbeat/HeartbeatServer.java
index a140b84..fab2dfc 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/heartbeat/HeartbeatServer.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/heartbeat/HeartbeatServer.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.cluster.config.ClusterConfig;
import org.apache.iotdb.cluster.config.ClusterDescriptor;
import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.utils.ClusterUtils;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.utils.CommonUtils;
import org.apache.iotdb.rpc.RpcTransportFactory;
@@ -79,6 +80,7 @@ public abstract class HeartbeatServer {
thisNode.setIp(config.getInternalIp());
thisNode.setMetaPort(config.getInternalMetaPort());
thisNode.setDataPort(config.getInternalDataPort());
+
thisNode.setClientIp(IoTDBDescriptor.getInstance().getConfig().getRpcAddress());
}
HeartbeatServer(Node thisNode) {
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
index 3ac46f3..23acce0 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
@@ -717,7 +717,8 @@ public class DataGroupMember extends RaftMember {
TSStatus result = forwardPlan(plan, leader.get(), getHeader());
Timer.Statistic.DATA_GROUP_MEMBER_FORWARD_PLAN.calOperationCostTimeFromStart(startTime);
if (!StatusUtils.NO_LEADER.equals(result)) {
- result.setRedirectNode(new EndPoint(leader.get().getIp(),
leader.get().getClientPort()));
+ result.setRedirectNode(
+ new EndPoint(leader.get().getClientIp(),
leader.get().getClientPort()));
return result;
}
}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java
index 194ba4c..4f945e0 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java
@@ -30,7 +30,13 @@ import
org.apache.iotdb.cluster.client.sync.SyncMetaHeartbeatClient;
import org.apache.iotdb.cluster.config.ClusterConstant;
import org.apache.iotdb.cluster.config.ClusterDescriptor;
import org.apache.iotdb.cluster.coordinator.Coordinator;
-import org.apache.iotdb.cluster.exception.*;
+import org.apache.iotdb.cluster.exception.AddSelfException;
+import org.apache.iotdb.cluster.exception.ConfigInconsistentException;
+import org.apache.iotdb.cluster.exception.EmptyIntervalException;
+import org.apache.iotdb.cluster.exception.LogExecutionException;
+import org.apache.iotdb.cluster.exception.PartitionTableUnavailableException;
+import org.apache.iotdb.cluster.exception.SnapshotInstallationException;
+import org.apache.iotdb.cluster.exception.StartUpCheckFailureException;
import org.apache.iotdb.cluster.log.Log;
import org.apache.iotdb.cluster.log.LogApplier;
import org.apache.iotdb.cluster.log.applier.MetaLogApplier;
@@ -44,9 +50,22 @@ import org.apache.iotdb.cluster.partition.PartitionGroup;
import org.apache.iotdb.cluster.partition.PartitionTable;
import org.apache.iotdb.cluster.partition.slot.SlotPartitionTable;
import org.apache.iotdb.cluster.query.ClusterPlanRouter;
-import org.apache.iotdb.cluster.rpc.thrift.*;
+import org.apache.iotdb.cluster.rpc.thrift.AddNodeResponse;
+import org.apache.iotdb.cluster.rpc.thrift.AppendEntryRequest;
+import org.apache.iotdb.cluster.rpc.thrift.CheckStatusResponse;
+import org.apache.iotdb.cluster.rpc.thrift.HeartBeatRequest;
+import org.apache.iotdb.cluster.rpc.thrift.HeartBeatResponse;
+import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.cluster.rpc.thrift.SendSnapshotRequest;
+import org.apache.iotdb.cluster.rpc.thrift.StartUpStatus;
+import org.apache.iotdb.cluster.rpc.thrift.TSMetaService;
import org.apache.iotdb.cluster.rpc.thrift.TSMetaService.AsyncClient;
-import org.apache.iotdb.cluster.server.*;
+import org.apache.iotdb.cluster.server.ClientServer;
+import org.apache.iotdb.cluster.server.DataClusterServer;
+import org.apache.iotdb.cluster.server.HardLinkCleaner;
+import org.apache.iotdb.cluster.server.NodeCharacter;
+import org.apache.iotdb.cluster.server.RaftServer;
+import org.apache.iotdb.cluster.server.Response;
import org.apache.iotdb.cluster.server.handlers.caller.AppendGroupEntryHandler;
import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler;
import org.apache.iotdb.cluster.server.handlers.caller.NodeStatusHandler;
@@ -83,12 +102,35 @@ import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.*;
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Paths;
-import java.util.*;
-import java.util.concurrent.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -423,7 +465,7 @@ public class MetaGroupMember extends RaftMember {
try {
if (logger.isInfoEnabled()) {
NodeReport report = genNodeReport();
- logger.debug(report.toString());
+ logger.info(report.toString());
}
} catch (Exception e) {
logger.error("{} exception occurred when generating node report", name,
e);
@@ -650,9 +692,10 @@ public class MetaGroupMember extends RaftMember {
public void processValidHeartbeatResp(HeartBeatResponse response, Node
receiver) {
// register the id of the node
if (response.isSetFollowerIdentifier()) {
- registerNodeIdentifier(receiver, response.getFollowerIdentifier());
+ registerNodeIdentifier(response.getFollower(),
response.getFollowerIdentifier());
// if all nodes' ids are known, we can build the partition table
if (allNodesIdKnown()) {
+ allNodes = new ArrayList<>(idNodeMap.values());
if (partitionTable == null) {
partitionTable = new SlotPartitionTable(allNodes, thisNode);
logger.info("Partition table is set up");
@@ -1344,7 +1387,8 @@ public class MetaGroupMember extends RaftMember {
} else if (!ClusterConstant.EMPTY_NODE.equals(leader.get())) {
TSStatus result = forwardPlan(plan, leader.get(), null);
if (!StatusUtils.NO_LEADER.equals(result)) {
- result.setRedirectNode(new EndPoint(leader.get().getIp(),
leader.get().getClientPort()));
+ result.setRedirectNode(
+ new EndPoint(leader.get().getClientIp(),
leader.get().getClientPort()));
return result;
}
}
@@ -1359,7 +1403,8 @@ public class MetaGroupMember extends RaftMember {
}
TSStatus result = forwardPlan(plan, leader.get(), null);
if (!StatusUtils.NO_LEADER.equals(result)) {
- result.setRedirectNode(new EndPoint(leader.get().getIp(),
leader.get().getClientPort()));
+ result.setRedirectNode(
+ new EndPoint(leader.get().getClientIp(),
leader.get().getClientPort()));
}
return result;
}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/service/BaseAsyncService.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/service/BaseAsyncService.java
index 565656a..07dbdea 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/service/BaseAsyncService.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/service/BaseAsyncService.java
@@ -154,7 +154,8 @@ public abstract class BaseAsyncService implements
RaftService.AsyncIface {
resultHandler.onComplete(
StatusUtils.getStatus(
status,
- new EndPoint(member.getThisNode().getIp(),
member.getThisNode().getClientPort())));
+ new EndPoint(
+ member.getThisNode().getClientIp(),
member.getThisNode().getClientPort())));
} catch (Exception e) {
resultHandler.onError(e);
}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/utils/ClusterNode.java
b/cluster/src/main/java/org/apache/iotdb/cluster/utils/ClusterNode.java
index 2284a93..6177197 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/utils/ClusterNode.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/utils/ClusterNode.java
@@ -32,8 +32,9 @@ public class ClusterNode extends Node {
// TODO hxd: maintain the client IP and Port
public ClusterNode() {}
- public ClusterNode(String ip, int metaPort, int nodeIdentifier, int
dataPort) {
- super(ip, metaPort, nodeIdentifier, dataPort);
+ public ClusterNode(
+ String ip, int metaPort, int nodeIdentifier, int dataPort, int
clientPort, String clientIp) {
+ super(ip, metaPort, nodeIdentifier, dataPort, clientPort, clientIp);
}
public ClusterNode(Node other) {
@@ -52,12 +53,13 @@ public class ClusterNode extends Node {
return Objects.equals(this.ip, that.ip)
&& this.dataPort == that.dataPort
&& this.metaPort == that.metaPort
- && this.clientPort == that.clientPort;
+ && this.clientPort == that.clientPort
+ && this.clientIp.equals(that.clientIp);
}
@Override
public int hashCode() {
- return Objects.hash(ip, metaPort, dataPort, clientPort);
+ return Objects.hash(ip, metaPort, dataPort, clientPort, clientIp);
}
@Override
@@ -74,6 +76,9 @@ public class ClusterNode extends Node {
+ dataPort
+ ", clientPort="
+ clientPort
+ + ", clientIp='"
+ + clientIp
+ + '\''
+ "}";
}
}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/utils/ClusterUtils.java
b/cluster/src/main/java/org/apache/iotdb/cluster/utils/ClusterUtils.java
index 315e730..c97ee9d 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/utils/ClusterUtils.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/utils/ClusterUtils.java
@@ -75,6 +75,8 @@ public class ClusterUtils {
*/
public static final int META_HEARTBEAT_PORT_OFFSET = 1;
+ public static String UNKNOWN_CLIENT_IP = "UNKNOWN_IP";
+
private ClusterUtils() {
// util class
}
@@ -303,34 +305,32 @@ public class ClusterUtils {
int idLastPos = str.indexOf(',', idFirstPos);
int dataPortFirstPos = str.indexOf("dataPort:", idLastPos) +
"dataPort:".length();
int dataPortLastPos = str.indexOf(',', dataPortFirstPos);
- int clientPortFirstPos = str.indexOf("clientPort:", idLastPos) +
"clientPort:".length();
- int clientPortLastPos = str.indexOf(')', clientPortFirstPos);
+ int clientPortFirstPos = str.indexOf("clientPort:", dataPortLastPos) +
"clientPort:".length();
+ int clientPortLastPos = str.indexOf(',', clientPortFirstPos);
+ int clientIpFirstPos = str.indexOf("clientIp:", clientPortLastPos) +
"clientIp:".length();
+ int clientIpLastPos = str.indexOf(')', clientIpFirstPos);
String ip = str.substring(ipFirstPos, ipLastPos);
int metaPort = Integer.parseInt(str.substring(metaPortFirstPos,
metaPortLastPos));
int id = Integer.parseInt(str.substring(idFirstPos, idLastPos));
int dataPort = Integer.parseInt(str.substring(dataPortFirstPos,
dataPortLastPos));
int clientPort = Integer.parseInt(str.substring(clientPortFirstPos,
clientPortLastPos));
+ String clientIp = str.substring(clientIpFirstPos, clientIpLastPos);
// TODO hxd: we do not set values to all fields of a Node.
- return new Node(ip, metaPort, id, dataPort).setClientPort(clientPort);
+ return new Node(ip, metaPort, id, dataPort, clientPort, clientIp);
}
public static Node parseNode(String nodeUrl) {
Node result = new Node();
String[] split = nodeUrl.split(":");
- if (split.length != 4) {
+ if (split.length != 2) {
logger.warn("Bad seed url: {}", nodeUrl);
return null;
}
String ip = split[0];
try {
int metaPort = Integer.parseInt(split[1]);
- int dataPort = Integer.parseInt(split[2]);
- int clientPort = Integer.parseInt(split[3]);
- result.setIp(ip);
- result.setMetaPort(metaPort);
- result.setDataPort(dataPort);
- result.setClientPort(clientPort);
+ result.setIp(ip).setMetaPort(metaPort).setClientIp(UNKNOWN_CLIENT_IP);
} catch (NumberFormatException e) {
logger.warn("Bad seed url: {}", nodeUrl);
}
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/common/TestPartitionedLogManager.java
b/cluster/src/test/java/org/apache/iotdb/cluster/common/TestPartitionedLogManager.java
index d026819..52681d2 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/common/TestPartitionedLogManager.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/common/TestPartitionedLogManager.java
@@ -32,9 +32,7 @@ public class TestPartitionedLogManager extends
PartitionedSnapshotLogManager {
super(
new TestLogApplier(),
null,
- new Node("localhost", 30001, 1, 40001)
- .setClientIp("localhost")
- .setClientPort(Constants.RPC_PORT),
+ new Node("localhost", 30001, 1, Constants.RPC_PORT, 6667, "localhost"),
null,
null,
null);
@@ -46,9 +44,7 @@ public class TestPartitionedLogManager extends
PartitionedSnapshotLogManager {
logApplier,
partitionTable,
header,
- new Node("localhost", 30001, 1, 40001)
- .setClientIp("localhost")
- .setClientPort(Constants.RPC_PORT),
+ new Node("localhost", 30001, 1, 40001, Constants.RPC_PORT,
"localhost"),
factory,
null);
}
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/common/TestUtils.java
b/cluster/src/test/java/org/apache/iotdb/cluster/common/TestUtils.java
index 104035a..1244502 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/common/TestUtils.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/common/TestUtils.java
@@ -76,6 +76,8 @@ public class TestUtils {
node.setMetaPort(ClusterDescriptor.getInstance().getConfig().getInternalMetaPort());
node.setDataPort(ClusterDescriptor.getInstance().getConfig().getInternalDataPort());
node.setNodeIdentifier(nodeNum);
+ node.setClientPort(IoTDBDescriptor.getInstance().getConfig().getRpcPort());
+
node.setClientIp(IoTDBDescriptor.getInstance().getConfig().getRpcAddress());
return node;
}
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/log/HardStateTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/log/HardStateTest.java
index 598d0df..5ee3f89 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/log/HardStateTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/log/HardStateTest.java
@@ -35,10 +35,7 @@ public class HardStateTest {
// Not NULL
HardState state = new HardState();
state.setCurrentTerm(2);
- state.setVoteFor(
- new Node("127.0.0.1", 30000, 0, 40000)
- .setClientIp("127.0.0.1")
- .setClientPort(Constants.RPC_PORT));
+ state.setVoteFor(new Node("127.0.0.1", 30000, 0, 40000,
Constants.RPC_PORT, "127.0.0.1"));
ByteBuffer buffer = state.serialize();
HardState newState = HardState.deserialize(buffer);
assertEquals(state, newState);
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/MetaLogApplierTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/MetaLogApplierTest.java
index b4a3949..dcfec31 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/MetaLogApplierTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/MetaLogApplierTest.java
@@ -83,9 +83,7 @@ public class MetaLogApplierTest extends IoTDBTest {
throws QueryProcessException, StorageGroupNotSetException,
StorageEngineException {
nodes.clear();
// TODO hxd:
- Node node = new Node("localhost", 1111, 0, 2222);
- node.setClientIp("localhost");
- node.setClientPort(Constants.RPC_PORT);
+ Node node = new Node("localhost", 1111, 0, 2222, Constants.RPC_PORT,
"localhost");
AddNodeLog log = new AddNodeLog();
log.setNewNode(node);
applier.apply(log);
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/log/logtypes/SerializeLogTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/log/logtypes/SerializeLogTest.java
index 963a3ae..bd64486 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/log/logtypes/SerializeLogTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/log/logtypes/SerializeLogTest.java
@@ -104,9 +104,7 @@ public class SerializeLogTest {
log.setCurrLogIndex(2);
log.setCurrLogTerm(2);
log.setNewNode(
- new Node("apache.iotdb.com", 1234, 1, 4321)
- .setClientIp("apache.iotdb.com")
- .setClientPort(Constants.RPC_PORT));
+ new Node("apache.iotdb.com", 1234, 1, 4321, Constants.RPC_PORT,
"apache.iotdb.com"));
ByteBuffer byteBuffer = log.serialize();
Log logPrime = LogParser.getINSTANCE().parse(byteBuffer);
assertEquals(log, logPrime);
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/partition/SlotPartitionTableTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/partition/SlotPartitionTableTest.java
index 8cc3f23..79e5eab 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/partition/SlotPartitionTableTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/partition/SlotPartitionTableTest.java
@@ -40,8 +40,17 @@ import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
-import org.apache.iotdb.db.qp.physical.sys.*;
+import org.apache.iotdb.db.qp.physical.sys.AuthorPlan;
+import org.apache.iotdb.db.qp.physical.sys.CountPlan;
+import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan;
+import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan;
+import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan;
import
org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan.LoadConfigurationPlanType;
+import org.apache.iotdb.db.qp.physical.sys.OperateFilePlan;
+import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
+import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan;
+import org.apache.iotdb.db.qp.physical.sys.ShowChildPathsPlan;
import org.apache.iotdb.db.qp.physical.sys.ShowPlan.ShowContentType;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -49,7 +58,11 @@ import
org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.junit.*;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
import org.mockito.internal.util.reflection.Whitebox;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,10 +71,21 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
import java.util.stream.IntStream;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
@Ignore // need maintenance
@SuppressWarnings({"java:S2699"})
@@ -181,20 +205,27 @@ public class SlotPartitionTableTest {
private void assertGetHeaderGroup(int start, int last) {
PartitionGroup group =
localTable.getHeaderGroup(
- new Node("localhost", 30000 + start, start, 40000 + start)
- .setClientIp("localhost")
- .setClientPort(Constants.RPC_PORT + start));
+ new Node(
+ "localhost",
+ 30000 + start,
+ start,
+ 40000 + start,
+ Constants.RPC_PORT + start,
+ "localhost"));
assertEquals(replica_size, group.size());
assertEquals(
- new Node("localhost", 30000 + start, start, 40000 + start)
- .setClientIp("localhost")
- .setClientPort(Constants.RPC_PORT + start),
+ new Node(
+ "localhost",
+ 30000 + start,
+ start,
+ 40000 + start,
+ Constants.RPC_PORT + start,
+ "localhost"),
group.getHeader());
assertEquals(
- new Node("localhost", 30000 + last, last, 40000 + last)
- .setClientIp("localhost")
- .setClientPort(Constants.RPC_PORT + start),
+ new Node(
+ "localhost", 30000 + last, last, 40000 + last, Constants.RPC_PORT
+ start, "localhost"),
group.get(replica_size - 1));
}
@@ -519,9 +550,7 @@ public class SlotPartitionTableTest {
}
private Node getNode(int i) {
- return new Node("localhost", 30000 + i, i, 40000 + i)
- .setClientIp("localhost")
- .setClientPort(Constants.RPC_PORT + i);
+ return new Node("localhost", 30000 + i, i, 40000 + i, Constants.RPC_PORT +
i, "localhost");
}
@Test
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/server/handlers/caller/HeartbeatHandlerTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/server/handlers/caller/HeartbeatHandlerTest.java
index b2290d4..e5baef2 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/server/handlers/caller/HeartbeatHandlerTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/server/handlers/caller/HeartbeatHandlerTest.java
@@ -36,7 +36,9 @@ import org.junit.Test;
import java.io.IOException;
-import static junit.framework.TestCase.*;
+import static junit.framework.TestCase.assertEquals;
+import static junit.framework.TestCase.assertFalse;
+import static junit.framework.TestCase.assertTrue;
public class HeartbeatHandlerTest {
@@ -74,9 +76,7 @@ public class HeartbeatHandlerTest {
response.setTerm(Response.RESPONSE_AGREE);
response.setLastLogTerm(-2);
response.setFollower(
- new Node("192.168.0.6", 9003, 6, 40010)
- .setClientIp("192.168.0.6")
- .setClientPort(Constants.RPC_PORT));
+ new Node("192.168.0.6", 9003, 6, 40010, Constants.RPC_PORT,
"192.168.0.6"));
catchUpFlag = false;
for (int i = 0; i < looseInconsistentNum; i++) {
handler.onComplete(response);
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java
index cea12ba..8cab2c1 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java
@@ -331,14 +331,8 @@ public class DataGroupMemberTest extends MemberTest {
testMetaMember.getTerm().set(10);
List<Log> metaLogs = TestUtils.prepareTestLogs(6);
metaLogManager.append(metaLogs);
- Node voteFor =
- new Node("127.0.0.1", 30000, 0, 40000)
- .setClientIp("127.0.0.1")
- .setClientPort(Constants.RPC_PORT);
- Node elector =
- new Node("127.0.0.1", 30001, 1, 40001)
- .setClientIp("127.0.0.1")
- .setClientPort(Constants.RPC_PORT + 1);
+ Node voteFor = new Node("127.0.0.1", 30000, 0, 40000, Constants.RPC_PORT,
"127.0.0.1");
+ Node elector = new Node("127.0.0.1", 30001, 1, 40001, Constants.RPC_PORT +
1, "127.0.0.1");
// a request with smaller term
ElectionRequest electionRequest = new ElectionRequest();
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
index 8600233..2eb30b1 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
@@ -1039,10 +1039,7 @@ public class MetaGroupMemberTest extends MemberTest {
request.setLeaderCommit(0);
request.setPrevLogIndex(-1);
request.setPrevLogTerm(-1);
- request.setLeader(
- new Node("127.0.0.1", 30000, 0, 40000)
- .setClientIp("127.0.0.1")
- .setClientPort(Constants.RPC_PORT));
+ request.setLeader(new Node("127.0.0.1", 30000, 0, 40000,
Constants.RPC_PORT, "127.0.0.1"));
AtomicReference<Long> result = new AtomicReference<>();
GenericHandler<Long> handler = new GenericHandler<>(TestUtils.getNode(0),
result);
new MetaAsyncService(testMetaMember).appendEntry(request, handler);
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/utils/Constants.java
b/cluster/src/test/java/org/apache/iotdb/cluster/utils/Constants.java
index 967052a..20b514b 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/utils/Constants.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/utils/Constants.java
@@ -20,6 +20,5 @@
package org.apache.iotdb.cluster.utils;
public class Constants {
-
public static int RPC_PORT = 6667;
}
diff --git a/cluster/src/test/resources/node1conf/iotdb-cluster.properties
b/cluster/src/test/resources/node1conf/iotdb-cluster.properties
index 76573d3..55ef6ab 100644
--- a/cluster/src/test/resources/node1conf/iotdb-cluster.properties
+++ b/cluster/src/test/resources/node1conf/iotdb-cluster.properties
@@ -20,7 +20,7 @@ internal_ip=127.0.0.1
internal_meta_port=9003
internal_data_port=40010
seed_nodes=127.0.0.1:9003:40010:6667,127.0.0.1:9005:40012:6668,127.0.0.1:9007:40014:6669
-default_replica_num=2
+default_replica_num=3
consistency_level=mid
connection_timeout_ms=20000
write_operation_timeout_ms=30000
diff --git a/cluster/src/test/resources/node2conf/iotdb-cluster.properties
b/cluster/src/test/resources/node2conf/iotdb-cluster.properties
index 68be6cb..1ca3be1 100644
--- a/cluster/src/test/resources/node2conf/iotdb-cluster.properties
+++ b/cluster/src/test/resources/node2conf/iotdb-cluster.properties
@@ -20,7 +20,7 @@ internal_ip=127.0.0.1
internal_meta_port=9005
internal_data_port=40012
seed_nodes=127.0.0.1:9003:40010:6667,127.0.0.1:9005:40012:6668,127.0.0.1:9007:40014:6669
-default_replica_num=2
+default_replica_num=3
consistency_level=mid
connection_timeout_ms=20000
write_operation_timeout_ms=30000
diff --git a/cluster/src/test/resources/node3conf/iotdb-cluster.properties
b/cluster/src/test/resources/node3conf/iotdb-cluster.properties
index 65184e7..ff4abca 100644
--- a/cluster/src/test/resources/node3conf/iotdb-cluster.properties
+++ b/cluster/src/test/resources/node3conf/iotdb-cluster.properties
@@ -20,7 +20,7 @@ internal_ip=127.0.0.1
internal_meta_port=9007
internal_data_port=40014
seed_nodes=127.0.0.1:9003:40010:6667,127.0.0.1:9005:40012:6668,127.0.0.1:9007:40014:6669
-default_replica_num=2
+default_replica_num=3
consistency_level=mid
connection_timeout_ms=20000
write_operation_timeout_ms=30000
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/SerializeUtils.java
b/server/src/main/java/org/apache/iotdb/db/utils/SerializeUtils.java
index f93137d..829f399 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/SerializeUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/SerializeUtils.java
@@ -121,27 +121,34 @@ public class SerializeUtils {
public static void serialize(Node node, DataOutputStream dataOutputStream) {
try {
- byte[] ipBytes = node.ip.getBytes();
- dataOutputStream.writeInt(ipBytes.length);
- dataOutputStream.write(ipBytes);
+ byte[] internalIpBytes = node.ip.getBytes();
+ dataOutputStream.writeInt(internalIpBytes.length);
+ dataOutputStream.write(internalIpBytes);
dataOutputStream.writeInt(node.metaPort);
dataOutputStream.writeInt(node.nodeIdentifier);
dataOutputStream.writeInt(node.dataPort);
dataOutputStream.writeInt(node.clientPort);
+ byte[] clientIpBytes = node.ip.getBytes();
+ dataOutputStream.writeInt(clientIpBytes.length);
+ dataOutputStream.write(clientIpBytes);
} catch (IOException e) {
// unreachable
}
}
public static void deserialize(Node node, ByteBuffer buffer) {
- int ipLength = buffer.getInt();
- byte[] ipBytes = new byte[ipLength];
- buffer.get(ipBytes);
- node.setIp(new String(ipBytes));
+ int internalIpLength = buffer.getInt();
+ byte[] internalIpBytes = new byte[internalIpLength];
+ buffer.get(internalIpBytes);
+ node.setIp(new String(internalIpBytes));
node.setMetaPort(buffer.getInt());
node.setNodeIdentifier(buffer.getInt());
node.setDataPort(buffer.getInt());
node.setClientPort(buffer.getInt());
+ int clientIpLength = buffer.getInt();
+ byte[] clientIpBytes = new byte[clientIpLength];
+ buffer.get(clientIpBytes);
+ node.setClientIp(new String(clientIpBytes));
}
public static void deserialize(Node node, DataInputStream stream) throws
IOException {
diff --git
a/server/src/test/java/org/apache/iotdb/db/utils/SerializeUtilsTest.java
b/server/src/test/java/org/apache/iotdb/db/utils/SerializeUtilsTest.java
index 79a85ad..ba91d9f 100644
--- a/server/src/test/java/org/apache/iotdb/db/utils/SerializeUtilsTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/utils/SerializeUtilsTest.java
@@ -86,12 +86,12 @@ public class SerializeUtilsTest {
@Test
public void serdesNodeTest() {
- Node node = new Node("127.0.0.1", 6667, 1, 6535).setClientPort(4678);
+ Node node = new Node("127.0.0.1", 6667, 1, 6535, 4678, "127.0.0.1");
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream outputStream = new DataOutputStream(baos);
SerializeUtils.serialize(node, outputStream);
ByteBuffer buffer = ByteBuffer.wrap(baos.toByteArray());
- Node anotherNode = new Node("127.0.0.1", 6667, 1,
6535).setClientPort(4678);
+ Node anotherNode = new Node("127.0.0.1", 6667, 1, 6535, 4678, "127.0.0.1");
SerializeUtils.deserialize(anotherNode, buffer);
Assert.assertEquals(node, anotherNode);
}
diff --git a/thrift/src/main/thrift/cluster.thrift
b/thrift/src/main/thrift/cluster.thrift
index 0defd44..0c4ebb5 100644
--- a/thrift/src/main/thrift/cluster.thrift
+++ b/thrift/src/main/thrift/cluster.thrift
@@ -113,8 +113,9 @@ struct Node {
2: required int metaPort
3: required int nodeIdentifier
4: required int dataPort
- 5: optional int clientPort
- 6: optional string clientIp
+ 5: required int clientPort
+ // When the cluster is set up for the first time, the clientIp of other
nodes is unknown
+ 6: required string clientIp
}
// leader -> follower
