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 2020/04/23 11:31:54 UTC
[GitHub] [incubator-iotdb] mychaow opened a new pull request #1090: [IOTDB-604] support more robust remote reader
mychaow opened a new pull request #1090:
URL: https://github.com/apache/incubator-iotdb/pull/1090
----------------------------------------------------------------
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] [incubator-iotdb] mychaow commented on a change in pull request #1090: [IOTDB-604] support more robust remote reader
Posted by GitBox <gi...@apache.org>.
mychaow commented on a change in pull request #1090:
URL: https://github.com/apache/incubator-iotdb/pull/1090#discussion_r415182791
##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.cluster.query.reader;
+
+import org.apache.iotdb.cluster.client.DataClient;
+import org.apache.iotdb.cluster.partition.PartitionGroup;
+import org.apache.iotdb.cluster.query.RemoteQueryContext;
+import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.cluster.rpc.thrift.SingleSeriesQueryRequest;
+import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler;
+import org.apache.iotdb.cluster.server.member.MetaGroupMember;
+import org.apache.iotdb.cluster.utils.SerializeUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.filter.TimeFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
+import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.iotdb.cluster.server.RaftServer.connectionTimeoutInMS;
+
+/**
+ * provide client which could connect to all nodes of the partitionGroup
+ */
+public class DataSourceInfo {
+ private static final Logger logger = LoggerFactory.getLogger(DataSourceInfo.class);
+
+ private long readerId;
+ private Node curSource;
+ private PartitionGroup partitionGroup;
+ private TSDataType dataType;
+ private SingleSeriesQueryRequest request;
+ private RemoteQueryContext context;
+ private MetaGroupMember metaGroupMember;
+ private List<Node> nodes;
+ private DataClient curClient;
+ private boolean isNoData;
+
+ private Map<Node, Boolean> visitedNodes;
Review comment:
ok, I will change the code.
----------------------------------------------------------------
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] [incubator-iotdb] jt2594838 commented on a change in pull request #1090: [IOTDB-604] support more robust remote reader
Posted by GitBox <gi...@apache.org>.
jt2594838 commented on a change in pull request #1090:
URL: https://github.com/apache/incubator-iotdb/pull/1090#discussion_r414399663
##########
File path: cluster/src/test/java/org/apache/iotdb/cluster/query/reader/RemoteSeriesReaderByTimestampTest.java
##########
@@ -71,18 +74,93 @@ public void fetchSingleSeriesByTimestamp(Node header, long readerId, long time,
resultHandler.onComplete(ByteBuffer.wrap(byteArrayOutputStream.toByteArray()));
}).start();
}
+
+ @Override
+ public void querySingleSeriesByTimestamp(SingleSeriesQueryRequest request, AsyncMethodCallback<Long> resultHandler) throws TException {
+ resultHandler.onComplete(1L);
+ }
};
}
};
@Test
public void test() throws IOException {
- RemoteSeriesReaderByTimestamp reader = new RemoteSeriesReaderByTimestamp(0,
- TestUtils.getNode(1), TestUtils.getNode(0),
- metaGroupMember);
+ PartitionGroup group = new PartitionGroup();
+ group.add(TestUtils.getNode(0));
+ group.add(TestUtils.getNode(1));
+ group.add(TestUtils.getNode(2));
+
+ SingleSeriesQueryRequest request = new SingleSeriesQueryRequest();
+ RemoteQueryContext context = new RemoteQueryContext(1);
+
+ DataSourceInfo sourceInfo = new DataSourceInfo(group, TSDataType.DOUBLE,
+ request, context, metaGroupMember, group);
+ sourceInfo.nextDataClient(true, Long.MIN_VALUE);
+
+ RemoteSeriesReaderByTimestamp reader = new RemoteSeriesReaderByTimestamp(sourceInfo);
+
for (int i = 0; i < 100; i++) {
assertEquals(i * 1.0, reader.getValueInTimestamp(i));
}
assertNull(reader.getValueInTimestamp(101));
}
+
+ @Test
+ public void testFailedNode() throws IOException {
+ batchData = TestUtils.genBatchData(TSDataType.DOUBLE, 0, 100);
+ PartitionGroup group = new PartitionGroup();
+ group.add(TestUtils.getNode(0));
+ group.add(TestUtils.getNode(1));
+ group.add(TestUtils.getNode(2));
+
+ SingleSeriesQueryRequest request = new SingleSeriesQueryRequest();
+ RemoteQueryContext context = new RemoteQueryContext(1);
+
+ DataSourceInfo sourceInfo = new DataSourceInfo(group, TSDataType.DOUBLE,
+ request, context, metaGroupMember, group);
+ sourceInfo.nextDataClient(true, Long.MIN_VALUE);
+ RemoteSeriesReaderByTimestamp reader = new RemoteSeriesReaderByTimestamp(sourceInfo);
+
+ // normal read
+ assertEquals(TestUtils.getNode(0), sourceInfo.getCurrentNode());
+ for (int i = 0; i < 50; i++) {
+ assertEquals(i * 1.0, reader.getValueInTimestamp(i));
+ }
+
+ // a bad client, change to another node
+ DataClient badClient = new DataClient(null, null, TestUtils.getNode(0), null) {
+ @Override
+ public void fetchSingleSeriesByTimestamp(Node header, long readerId, long time,
+ AsyncMethodCallback<ByteBuffer> resultHandler) throws TException {
+ throw new TException("Good bye.");
+ }
+
+ @Override
+ public void querySingleSeriesByTimestamp(SingleSeriesQueryRequest request, AsyncMethodCallback<Long> resultHandler) throws TException {
+ resultHandler.onComplete(1L);
+ }
+ };
+
+ reader.setClientForTest(badClient);
+ for (int i = 50; i < 80; i++) {
+ assertEquals(i * 1.0, reader.getValueInTimestamp(i));
+ }
+ assertEquals(TestUtils.getNode(1), sourceInfo.getCurrentNode());
+
+ // a bad client, change to another node again
+ reader.setClientForTest(badClient);
+ for (int i = 80; i < 90; i++) {
+ assertEquals(i * 1.0, reader.getValueInTimestamp(i));
+ }
+ assertEquals(TestUtils.getNode(2), sourceInfo.getCurrentNode());
+
+ // all node failed
+ reader.setClientForTest(badClient);
+ try {
+ reader.getValueInTimestamp(90);
+ fail();
+ } catch (IOException e) {
+ //ignore
Review comment:
It would be better to check the information or cause of the exception and see if it is what you have expected.
##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.cluster.query.reader;
+
+import org.apache.iotdb.cluster.client.DataClient;
+import org.apache.iotdb.cluster.partition.PartitionGroup;
+import org.apache.iotdb.cluster.query.RemoteQueryContext;
+import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.cluster.rpc.thrift.SingleSeriesQueryRequest;
+import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler;
+import org.apache.iotdb.cluster.server.member.MetaGroupMember;
+import org.apache.iotdb.cluster.utils.SerializeUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.filter.TimeFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
+import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.iotdb.cluster.server.RaftServer.connectionTimeoutInMS;
+
+/**
+ * provide client which could connect to all nodes of the partitionGroup
+ */
+public class DataSourceInfo {
+ private static final Logger logger = LoggerFactory.getLogger(DataSourceInfo.class);
+
+ private long readerId;
+ private Node curSource;
+ private PartitionGroup partitionGroup;
+ private TSDataType dataType;
+ private SingleSeriesQueryRequest request;
+ private RemoteQueryContext context;
+ private MetaGroupMember metaGroupMember;
+ private List<Node> nodes;
+ private DataClient curClient;
+ private boolean isNoData;
+
+ private Map<Node, Boolean> visitedNodes;
Review comment:
Recording the failed nodes is one way, but there is a chance that previously down nodes will go back online when the query proceeds, or maybe there is just a temporary network partition that caused the node time out.
For example, the query reads data of [1, 100] from node1 and node2, when it is reading 10, node1 fails, but later when it is reading 90, node1 recovers but node2 fails and the current implementation will miss a chance to let the query go on. Such a situation is more common when the query is complicated and time-consuming.
So I advise that do not record the visited node, but every time `nextDataClient` is called, traverse `nodes` once from the current position and only return null when the traverse ends but no nodes respond.
----------------------------------------------------------------
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] [incubator-iotdb] mychaow commented on a change in pull request #1090: [IOTDB-604] support more robust remote reader
Posted by GitBox <gi...@apache.org>.
mychaow commented on a change in pull request #1090:
URL: https://github.com/apache/incubator-iotdb/pull/1090#discussion_r415182791
##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/query/reader/DataSourceInfo.java
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.cluster.query.reader;
+
+import org.apache.iotdb.cluster.client.DataClient;
+import org.apache.iotdb.cluster.partition.PartitionGroup;
+import org.apache.iotdb.cluster.query.RemoteQueryContext;
+import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.cluster.rpc.thrift.SingleSeriesQueryRequest;
+import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler;
+import org.apache.iotdb.cluster.server.member.MetaGroupMember;
+import org.apache.iotdb.cluster.utils.SerializeUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.filter.TimeFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
+import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.iotdb.cluster.server.RaftServer.connectionTimeoutInMS;
+
+/**
+ * provide client which could connect to all nodes of the partitionGroup
+ */
+public class DataSourceInfo {
+ private static final Logger logger = LoggerFactory.getLogger(DataSourceInfo.class);
+
+ private long readerId;
+ private Node curSource;
+ private PartitionGroup partitionGroup;
+ private TSDataType dataType;
+ private SingleSeriesQueryRequest request;
+ private RemoteQueryContext context;
+ private MetaGroupMember metaGroupMember;
+ private List<Node> nodes;
+ private DataClient curClient;
+ private boolean isNoData;
+
+ private Map<Node, Boolean> visitedNodes;
Review comment:
Ok, I will change the code.
----------------------------------------------------------------
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