This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch cluster_new
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/cluster_new by this push:
new 0c07a44 fix query clientPool leak
new 0e8f77c Merge pull request #1618 from
LebronAl/cluster_new_bug_clientpool_query_leak
0c07a44 is described below
commit 0c07a44478ec176d5fc028093eda07a51a546022
Author: LebronAl <[email protected]>
AuthorDate: Thu Aug 13 21:09:31 2020 +0800
fix query clientPool leak
---
.../iotdb/cluster/query/reader/DataSourceInfo.java | 40 ++++++++++------------
.../cluster/server/member/MetaGroupMember.java | 8 ++---
.../cluster/query/reader/DatasourceInfoTest.java | 7 ++--
.../reader/RemoteSeriesReaderByTimestampTest.java | 10 +++---
.../query/reader/RemoteSimpleSeriesReaderTest.java | 4 +--
.../handlers/caller/HeartbeatHandlerTest.java | 6 ++--
6 files changed, 38 insertions(+), 37 deletions(-)
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
index eabc6df..8f2255a 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
@@ -24,7 +24,6 @@ import java.util.List;
import org.apache.iotdb.cluster.client.async.AsyncDataClient;
import org.apache.iotdb.cluster.client.sync.SyncClientAdaptor;
import org.apache.iotdb.cluster.client.sync.SyncDataClient;
-import org.apache.iotdb.cluster.config.ClusterDescriptor;
import org.apache.iotdb.cluster.partition.PartitionGroup;
import org.apache.iotdb.cluster.query.RemoteQueryContext;
import org.apache.iotdb.cluster.rpc.thrift.Node;
@@ -52,9 +51,9 @@ public class DataSourceInfo {
private RemoteQueryContext context;
private MetaGroupMember metaGroupMember;
private List<Node> nodes;
- private boolean isNoData;
private int curPos;
- private boolean noClient = false;
+ private boolean isNoData = false;
+ private boolean isNoClient = false;
public DataSourceInfo(PartitionGroup group, TSDataType dataType,
SingleSeriesQueryRequest request, RemoteQueryContext context,
@@ -66,16 +65,15 @@ public class DataSourceInfo {
this.context = context;
this.metaGroupMember = metaGroupMember;
this.nodes = nodes;
- this.isNoData = false;
// set to the last node so after nextDataClient() is called it will scan
from the first node
this.curPos = nodes.size() - 1;
this.curSource = nodes.get(curPos);
}
- public AsyncDataClient nextDataClient(boolean byTimestamp, long timestamp) {
+ public boolean hasNextDataClient(boolean byTimestamp, long timestamp) {
if (this.nodes.isEmpty()) {
this.isNoData = false;
- return null;
+ return false;
}
int nextNodePos = (this.curPos + 1) % this.nodes.size();
@@ -96,14 +94,13 @@ public class DataSourceInfo {
this.readerId = newReaderId;
this.curSource = node;
this.curPos = nextNodePos;
- return this.metaGroupMember
- .getAsyncDataClient(node,
RaftServer.getReadOperationTimeoutMS());
+ return true;
} else {
// the id being -1 means there is no satisfying data on the remote
node, create an
// empty reader to reduce further communication
- this.noClient = true;
+ this.isNoClient = true;
this.isNoData = true;
- return null;
+ return false;
}
}
} catch (TException | IOException e) {
@@ -115,13 +112,13 @@ public class DataSourceInfo {
nextNodePos = (nextNodePos + 1) % this.nodes.size();
if (nextNodePos == this.curPos) {
// has iterate over all nodes
- noClient = true;
+ isNoClient = true;
break;
}
}
// all nodes are failed
this.isNoData = false;
- return null;
+ return false;
}
private Long applyForReaderId(AsyncDataClient client, boolean byTimestamp,
long timestamp)
@@ -152,17 +149,21 @@ public class DataSourceInfo {
}
public AsyncDataClient getCurAsyncClient(int timeout) throws IOException {
- return noClient ? null :
metaGroupMember.getAsyncDataClient(this.curSource, timeout);
+ return isNoClient ? null :
metaGroupMember.getAsyncDataClient(this.curSource, timeout);
}
public SyncDataClient getCurSyncClient(int timeout) throws IOException {
- return noClient ? null : metaGroupMember.getSyncDataClient(this.curSource,
timeout);
+ return isNoClient ? null :
metaGroupMember.getSyncDataClient(this.curSource, timeout);
}
public boolean isNoData() {
return this.isNoData;
}
+ public boolean isNoClient(){
+ return this.isNoClient;
+ }
+
@Override
public String toString() {
return "DataSourceInfo{" +
@@ -180,10 +181,7 @@ public class DataSourceInfo {
* @throws IOException if all clients are unavailable.
*/
boolean checkCurClient() throws IOException {
- if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()
- && getCurAsyncClient(RaftServer.getConnectionTimeoutInMS()) == null ||
- !ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()
- && getCurSyncClient(RaftServer.getConnectionTimeoutInMS()) ==
null) {
+ if (isNoClient()) {
if (!isNoData()) {
throw new IOException("no available client.");
} else {
@@ -196,9 +194,9 @@ public class DataSourceInfo {
}
boolean switchNode(boolean byTimestamp, long timeOffset) throws IOException {
- AsyncDataClient newClient = nextDataClient(byTimestamp, timeOffset);
- logger.info("Client failed, changed to {}", newClient);
- if (newClient == null) {
+ boolean hasClient = hasNextDataClient(byTimestamp, timeOffset);
+ logger.info("Client failed, changed to {}", curSource);
+ if (!hasClient) {
if (!isNoData()) {
throw new IOException("no available client.");
} else {
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 d664c1d..51ccd0b 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
@@ -2322,8 +2322,8 @@ public class MetaGroupMember extends RaftMember {
DataSourceInfo dataSourceInfo = new DataSourceInfo(partitionGroup,
dataType, request,
(RemoteQueryContext) context, this, partitionGroup);
- AsyncDataClient client = dataSourceInfo.nextDataClient(true,
Long.MIN_VALUE);
- if (client != null) {
+ boolean hasClient = dataSourceInfo.hasNextDataClient(true, Long.MIN_VALUE);
+ if (hasClient) {
return new RemoteSeriesReaderByTimestamp(dataSourceInfo);
} else if (dataSourceInfo.isNoData()) {
return new EmptyReader();
@@ -2672,8 +2672,8 @@ public class MetaGroupMember extends RaftMember {
DataSourceInfo dataSourceInfo = new DataSourceInfo(partitionGroup,
dataType, request,
(RemoteQueryContext) context, this, orderedNodes);
- AsyncDataClient client = dataSourceInfo.nextDataClient(false,
Long.MIN_VALUE);
- if (client != null) {
+ boolean hasClient = dataSourceInfo.hasNextDataClient(false,
Long.MIN_VALUE);
+ if (hasClient) {
return new RemoteSimpleSeriesReader(dataSourceInfo);
} else if (dataSourceInfo.isNoData()) {
// there is no satisfying data on the remote node
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/DatasourceInfoTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/DatasourceInfoTest.java
index eca6d97..d05c58a84 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/DatasourceInfoTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/DatasourceInfoTest.java
@@ -20,8 +20,7 @@
package org.apache.iotdb.cluster.query.reader;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertFalse;
import java.io.IOException;
import org.apache.iotdb.cluster.client.async.AsyncDataClient;
@@ -68,8 +67,8 @@ public class DatasourceInfoTest {
DataSourceInfo sourceInfo = new DataSourceInfo(group, TSDataType.DOUBLE,
request, context, metaGroupMember, group);
- AsyncDataClient client = sourceInfo.nextDataClient(false, Long.MIN_VALUE);
+ boolean hasClient = sourceInfo.hasNextDataClient(false, Long.MIN_VALUE);
- assertNull(client);
+ assertFalse(hasClient);
}
}
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSeriesReaderByTimestampTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSeriesReaderByTimestampTest.java
index aa78946..2011d40 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSeriesReaderByTimestampTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSeriesReaderByTimestampTest.java
@@ -19,6 +19,10 @@
package org.apache.iotdb.cluster.query.reader;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -42,8 +46,6 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import static org.junit.Assert.*;
-
public class RemoteSeriesReaderByTimestampTest {
private BatchData batchData = TestUtils.genBatchData(TSDataType.DOUBLE, 0,
100);
@@ -122,7 +124,7 @@ public class RemoteSeriesReaderByTimestampTest {
DataSourceInfo sourceInfo = new DataSourceInfo(group, TSDataType.DOUBLE,
request, context, metaGroupMember, group);
- sourceInfo.nextDataClient(true, Long.MIN_VALUE);
+ sourceInfo.hasNextDataClient(true, Long.MIN_VALUE);
RemoteSeriesReaderByTimestamp reader = new
RemoteSeriesReaderByTimestamp(sourceInfo);
@@ -146,7 +148,7 @@ public class RemoteSeriesReaderByTimestampTest {
DataSourceInfo sourceInfo = new DataSourceInfo(group, TSDataType.DOUBLE,
request, context, metaGroupMember, group);
long startTime = System.currentTimeMillis();
- sourceInfo.nextDataClient(true, Long.MIN_VALUE);
+ sourceInfo.hasNextDataClient(true, Long.MIN_VALUE);
RemoteSeriesReaderByTimestamp reader = new
RemoteSeriesReaderByTimestamp(sourceInfo);
long endTime = System.currentTimeMillis();
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSimpleSeriesReaderTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSimpleSeriesReaderTest.java
index 86071b3..8f66f8e 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSimpleSeriesReaderTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSimpleSeriesReaderTest.java
@@ -123,7 +123,7 @@ public class RemoteSimpleSeriesReaderTest {
DataSourceInfo sourceInfo = new DataSourceInfo(group, TSDataType.DOUBLE,
request, context, metaGroupMember, group);
- sourceInfo.nextDataClient(false, Long.MIN_VALUE);
+ sourceInfo.hasNextDataClient(false, Long.MIN_VALUE);
reader = new RemoteSimpleSeriesReader(sourceInfo);
@@ -153,7 +153,7 @@ public class RemoteSimpleSeriesReaderTest {
DataSourceInfo sourceInfo = new DataSourceInfo(group, TSDataType.DOUBLE,
request, context, metaGroupMember, group);
- sourceInfo.nextDataClient(false, Long.MIN_VALUE);
+ sourceInfo.hasNextDataClient(false, Long.MIN_VALUE);
reader = new RemoteSimpleSeriesReader(sourceInfo);
// normal read
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 ca6f860..bb4bef1 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
@@ -30,7 +30,6 @@ import org.apache.iotdb.cluster.common.TestUtils;
import org.apache.iotdb.cluster.rpc.thrift.HeartBeatResponse;
import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.server.Response;
-import org.apache.iotdb.cluster.server.handlers.caller.HeartbeatHandler;
import org.apache.iotdb.cluster.server.member.MetaGroupMember;
import org.junit.After;
import org.junit.Before;
@@ -40,6 +39,7 @@ public class HeartbeatHandlerTest {
private MetaGroupMember metaGroupMember;
private boolean catchUpFlag;
+ private long looseInconsistentNum = 5;
@Before
public void setUp() {
@@ -70,7 +70,9 @@ public class HeartbeatHandlerTest {
response.setLastLogTerm(-2);
response.setFollower(new Node("192.168.0.6", 9003, 6, 40010));
catchUpFlag = false;
- handler.onComplete(response);
+ for (int i = 0; i < looseInconsistentNum; i++) {
+ handler.onComplete(response);
+ }
assertTrue(catchUpFlag);
}
