You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2021/06/22 09:17:50 UTC
[GitHub] [iotdb] ijihang opened a new pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
ijihang opened a new pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434
While the cluster is running, the coordinator node may go down, which may cause the session to be unable to read or write properly.i add a session interface to connect multiple nodes.Users can now define a list that contains multiple servers that can be connected. When one of the servers goes down, try again. If it fails, connect to the next one. If all the nodes in the list fail, an exception will be thrown.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] ijihang commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
ijihang commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r661072553
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +748,28 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
- try {
- if (transport != null) {
- close();
- init(endPoint);
- flag = true;
- }
- } catch (Exception e) {
- try {
- Thread.sleep(Config.RETRY_INTERVAL_MS);
- } catch (InterruptedException e1) {
- logger.error("reconnect is interrupted.", e1);
- Thread.currentThread().interrupt();
+ if (transport != null) {
+ transport.close();
+ int currHostIndex = endPointList.indexOf(session.defaultEndPoint);
+ for (int j = currHostIndex; j < endPointList.size(); j++) {
Review comment:
If all the endpoints in the list fail, I will try to connect again with the head node, traversing the list RETRY_NUM times in total
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] neuyilan commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
neuyilan commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r656718436
##########
File path: session/src/main/java/org/apache/iotdb/session/Session.java
##########
@@ -240,6 +241,70 @@ public Session(
this.enableCacheLeader = enableCacheLeader;
}
+ public Session(List<String> nodeUrls, String username, String password) {
Review comment:
It's better using java doc to specify the URL format clearly
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +748,28 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
- try {
- if (transport != null) {
- close();
- init(endPoint);
- flag = true;
- }
- } catch (Exception e) {
- try {
- Thread.sleep(Config.RETRY_INTERVAL_MS);
- } catch (InterruptedException e1) {
- logger.error("reconnect is interrupted.", e1);
- Thread.currentThread().interrupt();
+ if (transport != null) {
+ transport.close();
+ int currHostIndex = endPointList.indexOf(session.defaultEndPoint);
+ for (int j = currHostIndex; j < endPointList.size(); j++) {
Review comment:
If `currHostIndex` is the last index of `endPointList`, in your implementation, it can not try the others nodes again.
##########
File path: session/src/main/java/org/apache/iotdb/session/Session.java
##########
@@ -240,6 +241,70 @@ public Session(
this.enableCacheLeader = enableCacheLeader;
}
+ public Session(List<String> nodeUrls, String username, String password) {
+ this(
+ nodeUrls,
+ username,
+ password,
+ Config.DEFAULT_FETCH_SIZE,
+ null,
+ Config.DEFAULT_INITIAL_BUFFER_CAPACITY,
+ Config.DEFAULT_MAX_FRAME_SIZE,
+ Config.DEFAULT_CACHE_LEADER_MODE);
+ }
+
+ public Session(
+ List<String> nodeUrls,
+ String username,
+ String password,
+ int fetchSize,
+ ZoneId zoneId,
+ int thriftDefaultBufferSize,
+ int thriftMaxFrameSize,
+ boolean enableCacheLeader) {
+ this.nodeUrls = nodeUrls;
+ this.username = username;
+ this.password = password;
+ this.fetchSize = fetchSize;
+ this.zoneId = zoneId;
+ this.thriftDefaultBufferSize = thriftDefaultBufferSize;
+ this.thriftMaxFrameSize = thriftMaxFrameSize;
+ this.enableCacheLeader = enableCacheLeader;
+ }
+
+ public synchronized void clusterOpen() throws IoTDBConnectionException {
Review comment:
+1, I don't think it's necessary to change the default behavior of current users using session, in other words, I think we just need to add one interface like[1], then the user can just call the following codes[2] like before;
[1] `public Session(List<String> nodeUrls, String username, String password) `
[2]
`
session= new Session(nodeUrls, "root", "root");
session.open(false);
session.setFetchSize(10000);
try {
session.setStorageGroup("root.sg1");
} catch (StatementExecutionException e) {
if (e.getStatusCode() != TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode()) {
throw e;
}
}
`
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionUtils.java
##########
@@ -149,4 +158,32 @@ private static void getValueBufferOfDataType(
String.format("Data type %s is not supported.", dataType));
}
}
+
+ public static List<EndPoint> parseSeedNodeUrls(List<String> nodeUrls) {
+ if (nodeUrls == null) {
+ return Collections.emptyList();
+ }
+ List<EndPoint> endPointsList = new ArrayList<>();
+ for (String nodeUrl : nodeUrls) {
+ EndPoint endPoint = parseNodeUrl(nodeUrl);
+ endPointsList.add(endPoint);
+ }
+ return endPointsList;
+ }
+
+ private static EndPoint parseNodeUrl(String nodeUrl) {
+ EndPoint endPoint = new EndPoint();
+ String[] split = nodeUrl.split(":");
+ if (split.length != 2) {
+ return null;
+ }
+ String ip = split[0];
+ try {
+ int rpcPort = Integer.parseInt(split[1]);
+ return endPoint.setIp(ip).setPort(rpcPort);
+ } catch (NumberFormatException e) {
+ logger.warn("parse url fail {}", nodeUrl);
+ }
+ return endPoint;
+ }
Review comment:
If the user given one wrong URL format, in the code implement, the `NumberFormatException` may occur, you will return one `empty` endPoint.
In my opinion, if the seed URL format error, It's better rethrow the exception to the user.
##########
File path: session/src/main/java/org/apache/iotdb/session/Session.java
##########
@@ -525,16 +584,12 @@ public SessionDataSet executeQueryStatement(String sql, long timeoutInMs)
private SessionDataSet executeStatementMayRedirect(String sql, long timeoutInMs)
throws StatementExecutionException, IoTDBConnectionException {
try {
- logger.info("{} execute sql {}", defaultSessionConnection.getEndPoint(), sql);
+ logger.info("{} execute sql {}", defaultEndPoint, sql);
return defaultSessionConnection.executeQueryStatement(sql, timeoutInMs);
} catch (RedirectException e) {
handleQueryRedirection(e.getEndPoint());
if (enableQueryRedirection) {
- logger.debug(
- "{} redirect query {} to {}",
- defaultSessionConnection.getEndPoint(),
- sql,
- e.getEndPoint());
+ logger.debug("{} redirect query {} to {}", defaultEndPoint, sql, e.getEndPoint());
Review comment:
the same as above
##########
File path: session/src/main/java/org/apache/iotdb/session/Session.java
##########
@@ -525,16 +584,12 @@ public SessionDataSet executeQueryStatement(String sql, long timeoutInMs)
private SessionDataSet executeStatementMayRedirect(String sql, long timeoutInMs)
throws StatementExecutionException, IoTDBConnectionException {
try {
- logger.info("{} execute sql {}", defaultSessionConnection.getEndPoint(), sql);
+ logger.info("{} execute sql {}", defaultEndPoint, sql);
Review comment:
`defaultSessionConnection.getEndPoint() `may not equal `defaultEndPoint` as long as we enable redirection
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionUtils.java
##########
@@ -149,4 +158,32 @@ private static void getValueBufferOfDataType(
String.format("Data type %s is not supported.", dataType));
}
}
+
+ public static List<EndPoint> parseSeedNodeUrls(List<String> nodeUrls) {
+ if (nodeUrls == null) {
+ return Collections.emptyList();
+ }
+ List<EndPoint> endPointsList = new ArrayList<>();
+ for (String nodeUrl : nodeUrls) {
+ EndPoint endPoint = parseNodeUrl(nodeUrl);
+ endPointsList.add(endPoint);
Review comment:
`parseNodeUrl` method may return `null`, we can not add one `null` endpoint to the `endPointsList`
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -145,6 +153,24 @@ private void init(EndPoint endPoint) throws IoTDBConnectionException {
}
}
+ private void initClusterConn() throws IoTDBConnectionException {
+ for (EndPoint endPoint : endPointList) {
+ if (endPoint == null) {
+ continue;
+ }
Review comment:
Its better not allow `null` endPoint added in the `endPointList` in the previous `parseSeedNodeUrls` method. In other words, abnormal judgment should be made as soon as possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] mychaow commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
mychaow commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r658569624
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +748,28 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
- try {
- if (transport != null) {
- close();
- init(endPoint);
- flag = true;
- }
- } catch (Exception e) {
- try {
- Thread.sleep(Config.RETRY_INTERVAL_MS);
- } catch (InterruptedException e1) {
- logger.error("reconnect is interrupted.", e1);
- Thread.currentThread().interrupt();
+ if (transport != null) {
+ transport.close();
+ int currHostIndex = endPointList.indexOf(session.defaultEndPoint);
+ for (int j = currHostIndex; j < endPointList.size(); j++) {
Review comment:
not right, can not try the previous endpoint.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] wangchao316 commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
wangchao316 commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r663898431
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +751,27 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
Review comment:
I think we should connect nodes randomly, so that we can reduce the pressure on one node.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] wangchao316 commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
wangchao316 commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r663898431
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +751,27 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
Review comment:
I think we should connect nodes randomly, so that we can reduce the pressure on one node.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] mychaow commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
mychaow commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r664632279
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +747,38 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
- try {
- if (transport != null) {
- close();
- init(endPoint);
- flag = true;
+ if (transport != null) {
+ transport.close();
+ int currHostIndex = endPointList.indexOf(endPoint);
+ if (currHostIndex == -1) {
+ logger.warn(
+ "endPoint:{} not in endPointList,Try the first one in the endPointList", endPoint);
+ currHostIndex = 0;
}
- } catch (Exception e) {
- try {
- Thread.sleep(Config.RETRY_INTERVAL_MS);
- } catch (InterruptedException e1) {
- logger.error("reconnect is interrupted.", e1);
- Thread.currentThread().interrupt();
+ int tag = 0;
Review comment:
tag->tryHostNum
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +747,38 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
Review comment:
flag -> connectedSuccess
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] mychaow merged pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
mychaow merged pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] ijihang commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
ijihang commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r656816433
##########
File path: session/src/main/java/org/apache/iotdb/session/Session.java
##########
@@ -240,6 +241,70 @@ public Session(
this.enableCacheLeader = enableCacheLeader;
}
+ public Session(List<String> nodeUrls, String username, String password) {
+ this(
+ nodeUrls,
+ username,
+ password,
+ Config.DEFAULT_FETCH_SIZE,
+ null,
+ Config.DEFAULT_INITIAL_BUFFER_CAPACITY,
+ Config.DEFAULT_MAX_FRAME_SIZE,
+ Config.DEFAULT_CACHE_LEADER_MODE);
+ }
+
+ public Session(
+ List<String> nodeUrls,
+ String username,
+ String password,
+ int fetchSize,
+ ZoneId zoneId,
+ int thriftDefaultBufferSize,
+ int thriftMaxFrameSize,
+ boolean enableCacheLeader) {
+ this.nodeUrls = nodeUrls;
+ this.username = username;
+ this.password = password;
+ this.fetchSize = fetchSize;
+ this.zoneId = zoneId;
+ this.thriftDefaultBufferSize = thriftDefaultBufferSize;
+ this.thriftMaxFrameSize = thriftMaxFrameSize;
+ this.enableCacheLeader = enableCacheLeader;
+ }
+
+ public synchronized void clusterOpen() throws IoTDBConnectionException {
Review comment:
ok,i modify in new commit,using the old open method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] ijihang commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
ijihang commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r665026510
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +751,27 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
Review comment:
ok, in new commit,connect nodes randomly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] mychaow commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
mychaow commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r656113582
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +751,27 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
Review comment:
Always try the first RETRY_NUM clients, it's not good. Maybe you can record a nextHostIndex to try next hostId.
##########
File path: session/src/main/java/org/apache/iotdb/session/Session.java
##########
@@ -240,6 +241,70 @@ public Session(
this.enableCacheLeader = enableCacheLeader;
}
+ public Session(List<String> nodeUrls, String username, String password) {
+ this(
+ nodeUrls,
+ username,
+ password,
+ Config.DEFAULT_FETCH_SIZE,
+ null,
+ Config.DEFAULT_INITIAL_BUFFER_CAPACITY,
+ Config.DEFAULT_MAX_FRAME_SIZE,
+ Config.DEFAULT_CACHE_LEADER_MODE);
+ }
+
+ public Session(
+ List<String> nodeUrls,
+ String username,
+ String password,
+ int fetchSize,
+ ZoneId zoneId,
+ int thriftDefaultBufferSize,
+ int thriftMaxFrameSize,
+ boolean enableCacheLeader) {
+ this.nodeUrls = nodeUrls;
+ this.username = username;
+ this.password = password;
+ this.fetchSize = fetchSize;
+ this.zoneId = zoneId;
+ this.thriftDefaultBufferSize = thriftDefaultBufferSize;
+ this.thriftMaxFrameSize = thriftMaxFrameSize;
+ this.enableCacheLeader = enableCacheLeader;
+ }
+
+ public synchronized void clusterOpen() throws IoTDBConnectionException {
Review comment:
I think it's no need to create a new open function, you could put the logic which just create SessionConnection who use the nodeUrl in the contructSessionConn.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] mychaow commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
mychaow commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r663589845
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -721,20 +746,28 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
- for (int i = 1; i <= Config.RETRY_NUM; i++) {
- try {
- if (transport != null) {
- close();
- init(endPoint);
- flag = true;
+ if (transport != null) {
+ transport.close();
+ int currHostIndex = endPointList.indexOf(session.defaultEndPoint);
Review comment:
currHostIndex maybe -1, so you should handle it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] ijihang commented on a change in pull request #3434: [IOTDB-1399]Add a session interface to connect multiple nodes
Posted by GitBox <gi...@apache.org>.
ijihang commented on a change in pull request #3434:
URL: https://github.com/apache/iotdb/pull/3434#discussion_r656815804
##########
File path: session/src/main/java/org/apache/iotdb/session/SessionConnection.java
##########
@@ -722,20 +751,27 @@ protected void testInsertTablets(TSInsertTabletsReq request)
private boolean reconnect() {
boolean flag = false;
for (int i = 1; i <= Config.RETRY_NUM; i++) {
Review comment:
in new commit,Instead of trying the first node in the list every time, try the current node again, and then try the next node in the list
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org