This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new b3f21a3 [IOTDB-832] fix reconnection failure bug in sessionPool
(#1610)
b3f21a3 is described below
commit b3f21a3ef677700f6d2f71ccc11c0db6a1ca45b2
Author: Xiangdong Huang <[email protected]>
AuthorDate: Tue Aug 11 12:14:13 2020 +0800
[IOTDB-832] fix reconnection failure bug in sessionPool (#1610)
* [iotdb-832] fix sessionPool logic when reconnection failed.
Co-authored-by: xiangdong huang <[email protected]>
Co-authored-by: qiaojialin <[email protected]>
---
docs/UserGuide/Client/Status Codes.md | 1 +
docs/zh/UserGuide/Client/Status Codes.md | 1 +
.../org/apache/iotdb/db/engine/StorageEngine.java | 4 +-
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 1 +
.../org/apache/iotdb/session/pool/SessionPool.java | 15 ++++-
.../apache/iotdb/session/pool/SessionPoolTest.java | 69 +++++++++++-----------
6 files changed, 52 insertions(+), 39 deletions(-)
diff --git a/docs/UserGuide/Client/Status Codes.md
b/docs/UserGuide/Client/Status Codes.md
index 55245a4..bb85281 100644
--- a/docs/UserGuide/Client/Status Codes.md
+++ b/docs/UserGuide/Client/Status Codes.md
@@ -66,6 +66,7 @@ Here is a list of Status Code and related message:
|314|TSFILE_PROCESSOR_ERROR|TsFile processor related error|
|315|PATH_ILLEGAL|Illegal path|
|316|LOAD_FILE_ERROR|Meet error while loading file|
+|317|STORAGE_GROUP_NOT_READY| The storage group is in recovery mode, not ready
fore accepting read/write operation|
|400|EXECUTE_STATEMENT_ERROR|Execute statement error|
|401|SQL_PARSE_ERROR|Meet error while parsing SQL|
|402|GENERATE_TIME_ZONE_ERROR|Meet error while generating time zone|
diff --git a/docs/zh/UserGuide/Client/Status Codes.md
b/docs/zh/UserGuide/Client/Status Codes.md
index f509e63..a1ca981 100644
--- a/docs/zh/UserGuide/Client/Status Codes.md
+++ b/docs/zh/UserGuide/Client/Status Codes.md
@@ -66,6 +66,7 @@ try {
|314|TSFILE_PROCESSOR_ERROR|TsFile处理器相关错误|
|315|PATH_ILLEGAL|路径不合法|
|316|LOAD_FILE_ERROR|加载文件错误|
+|317|STORAGE_GROUP_NOT_READY| 存储组还在恢复中，还不能接受读写操作 |
|400|EXECUTE_STATEMENT_ERROR|执行语句错误|
|401|SQL_PARSE_ERROR|SQL语句分析错误|
|402|GENERATE_TIME_ZONE_ERROR|生成时区错误|
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index a11e944..57f97ff 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -47,6 +47,7 @@ import org.apache.iotdb.db.service.ServiceType;
import org.apache.iotdb.db.utils.FilePathUtils;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.db.utils.UpgradeUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
import org.slf4j.Logger;
@@ -293,7 +294,8 @@ public class StorageEngine implements IService {
}
} else {
// not finished recover, refuse the request
- throw new StorageEngineException("the sg " + storageGroupName + "
may not ready now, please wait and retry later");
+ throw new StorageEngineException("the sg " + storageGroupName + "
may not ready now, please wait and retry later",
+ TSStatusCode.STORAGE_GROUP_NOT_READY.getStatusCode());
}
}
return processor;
diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index 08aa3ad..ba3fdb6 100644
--- a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -43,6 +43,7 @@ public enum TSStatusCode {
TSFILE_PROCESSOR_ERROR(314),
PATH_ILLEGAL(315),
LOAD_FILE_ERROR(316),
+ STORAGE_GROUP_NOT_READY(317),
EXECUTE_STATEMENT_ERROR(400),
SQL_PARSE_ERROR(401),
diff --git
a/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
b/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
index 4a03917..d50a12e 100644
--- a/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
+++ b/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
@@ -115,10 +115,12 @@ public class SessionPool {
//we have to wait for someone returns a session.
try {
this.wait(1000);
- if (System.currentTimeMillis() - start > 60_000) {
+ long time = timeout < 60_000 ? timeout : 60_000;
+ if (System.currentTimeMillis() - start > time) {
logger.warn(
"the SessionPool has wait for {} seconds to get a new
connection: {}:{} with {}, {}",
(System.currentTimeMillis() - start) / 1000, ip, port,
user, password);
+ logger.warn("current occupied size {}, queue size {},
considered size {} ",occupied.size(), queue.size(), size);
if (System.currentTimeMillis() - start > timeout) {
throw new IoTDBConnectionException(
String.format("timeout to get a connection from %s:%s",
ip, port));
@@ -139,7 +141,16 @@ public class SessionPool {
logger.debug("Create a new Session {}, {}, {}, {}", ip, port, user,
password);
}
session = new Session(ip, port, user, password, fetchSize);
- session.open(enableCompression);
+ try {
+ session.open(enableCompression);
+ } catch (IoTDBConnectionException e) {
+ //if exception, we will throw the exception.
+ //Meanwhile, we have to set size--
+ synchronized (this) {
+ size --;
+ }
+ throw e;
+ }
return session;
}
}
diff --git
a/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
b/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
index c35d044..af9bf2a 100644
--- a/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
+++ b/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
@@ -99,15 +99,7 @@ public class SessionPoolTest {
public void incorrectExecuteQueryStatement() {
SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3);
ExecutorService service = Executors.newFixedThreadPool(10);
- for (int i = 0; i < 10; i++) {
- try {
- pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i),
- Collections.singletonList(TSDataType.INT64),
- Collections.singletonList((long) i));
- } catch (IoTDBConnectionException | StatementExecutionException e) {
- fail();
- }
- }
+ write10Data(pool, true);
//now let's query
for (int i = 0; i < 10; i++) {
final int no = i;
@@ -143,15 +135,7 @@ public class SessionPoolTest {
private void correctQuery(SessionPool pool) {
ExecutorService service = Executors.newFixedThreadPool(10);
- for (int i = 0; i < 10; i++) {
- try {
- pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i),
- Collections.singletonList(TSDataType.INT64),
- Collections.singletonList((long) i));
- } catch (IoTDBConnectionException | StatementExecutionException e) {
- fail();
- }
- }
+ write10Data(pool, true);
//now let's query
for (int i = 0; i < 10; i++) {
final int no = i;
@@ -180,15 +164,7 @@ public class SessionPoolTest {
@Test
public void tryIfTheServerIsRestart() {
SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3,
1, 6000, false);
- for (int i = 0; i < 10; i++) {
- try {
- pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i),
- Collections.singletonList(TSDataType.INT64),
- Collections.singletonList((long) i));
- } catch (IoTDBConnectionException | StatementExecutionException e) {
- fail();
- }
- }
+ write10Data(pool, true);
SessionDataSetWrapper wrapper = null;
try {
wrapper = pool.executeQueryStatement("select * from root.sg1.d1 where
time > 1");
@@ -212,15 +188,7 @@ public class SessionPoolTest {
@Test
public void tryIfTheServerIsRestartButDataIsGotten() {
SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3,
1, 60000, false);
- for (int i = 0; i < 10; i++) {
- try {
- pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i),
- Collections.singletonList(TSDataType.INT64),
- Collections.singletonList((long) i));
- } catch (IoTDBConnectionException | StatementExecutionException e) {
- fail();
- }
- }
+ write10Data(pool, true);
assertEquals(1, pool.currentAvailableSize());
SessionDataSetWrapper wrapper = null;
try {
@@ -240,4 +208,33 @@ public class SessionPoolTest {
pool.close();
}
+ @Test
+ public void restart() throws Exception {
+ SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 1,
1, 1000, false);
+ write10Data(pool, true);
+ //stop the server.
+ EnvironmentUtils.stopDaemon();
+ //all this ten data will fail.
+ write10Data(pool, false);
+ //restart the server
+ EnvironmentUtils.reactiveDaemon();
+ write10Data(pool, true);
+
+ }
+
+ private void write10Data(SessionPool pool, boolean failWhenThrowException) {
+ for (int i = 0; i < 10; i++) {
+ try {
+ pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i),
+ Collections.singletonList(TSDataType.INT64),
+ Collections.singletonList((long) i));
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ //will fail this 10 times.
+ if (failWhenThrowException) {
+ fail();
+ }
+ }
+ }
+ }
+
}
\ No newline at end of file
