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 2022/03/11 06:27:54 UTC
[GitHub] [iotdb] Cpaulyz opened a new pull request #5200: [IOTDB-2700] Sync transfer adaption to receiver
Cpaulyz opened a new pull request #5200:
URL: https://github.com/apache/iotdb/pull/5200
Sync transfer adaption to receiver
--
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] irvine0109 commented on a change in pull request #5200: [To new_sync][IOTDB-2700] Sync transfer adaption to receiver
Posted by GitBox <gi...@apache.org>.
irvine0109 commented on a change in pull request #5200:
URL: https://github.com/apache/iotdb/pull/5200#discussion_r825259828
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
##########
@@ -170,31 +182,32 @@ public TransportStatus transportData(
}
if (type != Type.FILE) {
-
buff.position(pos);
int length = buff.capacity();
byte[] byteArray = new byte[length];
buff.get(byteArray);
- try (InputStream inputStream = new ByteArrayInputStream(byteArray);
- DataInputStream dataInputStream = new DataInputStream(inputStream)) {
- PipeData pipeData = PipeData.deserialize(dataInputStream);
- // Do with file
- // BufferedPipeDataQueue.offer(pipeData);
+ try {
+ PipeData pipeData = PipeData.deserialize(byteArray);
+ if (type == Type.TSFILE) {
+ // Do with file
+ handleTsFilePipeData((TsFilePipeData) pipeData, fileDir);
+ }
+ PipeDataQueueFactory.getBufferedPipeDataQueue(getPipeLogDirPath(identityInfo))
+ .offer(pipeData);
} catch (IOException | IllegalPathException e) {
e.printStackTrace();
Review comment:
It's my fault. May use the statement like `logger.error("Error XXX", e);` is appropriate. The statement like `e.printStackTrace();` may not print to logger.
##########
File path: server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java
##########
@@ -0,0 +1,193 @@
+/*
+ * 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.db.newsync.transport;
+
+import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
+import org.apache.iotdb.db.newsync.pipedata.DeletionPipeData;
+import org.apache.iotdb.db.newsync.pipedata.PipeData;
+import org.apache.iotdb.db.newsync.pipedata.SchemaPipeData;
+import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
+import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueue;
+import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueueFactory;
+import org.apache.iotdb.db.newsync.sender.pipe.Pipe;
+import org.apache.iotdb.db.newsync.sender.pipe.TsFilePipe;
+import org.apache.iotdb.db.newsync.transport.client.TransportClient;
+import org.apache.iotdb.db.newsync.transport.server.TransportServerManager;
+import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import org.apache.commons.io.FileUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class TransportServiceTest {
+ /** create tsfile and move to tmpDir for sync test */
+ File tmpDir = new File("target/synctest");
+
+ String pipeName1 = "pipe1";
+ String remoteIp1 = "127.0.0.1";
+ long createdTime1 = System.currentTimeMillis();
+ File fileDir = new File(SyncPathUtil.getReceiverFileDataDir(pipeName1, remoteIp1, createdTime1));
+ PipeDataQueue pipeDataQueue =
+ PipeDataQueueFactory.getBufferedPipeDataQueue(
+ SyncPathUtil.getReceiverPipeLogDir(pipeName1, remoteIp1, createdTime1));
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.envSetUp();
+ if (!tmpDir.exists()) {
+ tmpDir.mkdirs();
+ }
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ pipeDataQueue.clear();
+ FileUtils.deleteDirectory(tmpDir);
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void test() throws Exception {
+ // 1. prepare fake file
+ File tsfile = new File(tmpDir, "test.tsfile");
+ File resourceFile = new File(tsfile.getAbsoluteFile() + TsFileResource.RESOURCE_SUFFIX);
+ File modsFile = new File(tsfile.getAbsoluteFile() + ModificationFile.FILE_SUFFIX);
+ FileWriter out = new FileWriter(tsfile);
+ out.write("tsfile");
+ out.flush();
+ out.close();
+ out = new FileWriter(resourceFile);
+ out.write("resource");
+ out.flush();
+ out.close();
+ out = new FileWriter(modsFile);
+ out.write("mods");
+ out.flush();
+ out.close();
+
+ // 2. prepare pipelog and pipeDataQueue
+ int serialNum = 0;
+ List<PipeData> pipeDataList = new ArrayList<>();
+ pipeDataList.add(
+ new SchemaPipeData(new SetStorageGroupPlan(new PartialPath("root.vehicle")), serialNum++));
+ pipeDataList.add(
+ new SchemaPipeData(
+ new CreateTimeSeriesPlan(
+ new PartialPath("root.vehicle.d0.s0"),
+ new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.RLE)),
+ serialNum++));
+ TsFilePipeData tsFilePipeData = new TsFilePipeData(tsfile.getPath(), serialNum++);
+ pipeDataList.add(tsFilePipeData);
+ Deletion deletion = new Deletion(new PartialPath("root.vehicle.**"), 0, 33, 38);
+ pipeDataList.add(new DeletionPipeData(deletion, serialNum++));
+
+ // 3. start server
+ TransportServerManager.getInstance().startService();
+
+ // 4. start client
+ Pipe pipe = new TsFilePipe(createdTime1, pipeName1, null, 0, false);
+ TransportClient client = new TransportClient(pipe, "127.0.0.1", 5555);
+ for (PipeData pipeData : pipeDataList) {
+ client.senderTransport(pipeData);
+ }
+
+ // 5. check file
+ Thread.sleep(1000);
+ client.close();
+ TransportServerManager.getInstance().stopService();
+ File[] targetFiles = fileDir.listFiles((dir1, name) -> name.equals(tsfile.getName()));
+ Assert.assertNotNull(targetFiles);
+ Assert.assertEquals(1, targetFiles.length);
+ compareFile(targetFiles[0], tsfile);
+ File[] resourceFiles = fileDir.listFiles((dir1, name) -> name.equals(resourceFile.getName()));
+ Assert.assertNotNull(resourceFiles);
+ Assert.assertEquals(1, resourceFiles.length);
+ compareFile(resourceFiles[0], resourceFile);
+ File[] modsFiles = fileDir.listFiles((dir1, name) -> name.equals(modsFile.getName()));
+ Assert.assertNotNull(modsFiles);
+ Assert.assertEquals(1, modsFiles.length);
+ compareFile(modsFiles[0], modsFile);
+
+ // 6. check pipedata
+ tsFilePipeData.setTsFilePath(fileDir.getAbsolutePath() + File.separator + tsfile.getName());
+ ExecutorService es1 = Executors.newSingleThreadExecutor();
+ List<PipeData> resPipeData = new ArrayList<>();
+ es1.execute(
+ () -> {
+ for (int i = 0; i < pipeDataList.size(); i++) {
+ try {
+ resPipeData.add(pipeDataQueue.take());
+ pipeDataQueue.commit();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ es1.shutdownNow();
+ Assert.assertEquals(pipeDataList.size(), resPipeData.size());
+ for (int i = 0; i < resPipeData.size(); i++) {
+ Assert.assertEquals(pipeDataList.get(i), resPipeData.get(i));
+ }
+ }
+
+ private void compareFile(File firFile, File secFile) {
+ try {
+ BufferedInputStream fir = new BufferedInputStream(new FileInputStream(firFile));
+ BufferedInputStream sec = new BufferedInputStream(new FileInputStream(secFile));
+ // 比较文件的长度是否一样
Review comment:
// To compare the length of the files.
BTW, Why don't compare the hash of the files.
--
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] yschengzi commented on a change in pull request #5200: [To new_sync][IOTDB-2700] Sync transfer adaption to receiver
Posted by GitBox <gi...@apache.org>.
yschengzi commented on a change in pull request #5200:
URL: https://github.com/apache/iotdb/pull/5200#discussion_r825745694
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
##########
@@ -218,7 +220,6 @@ public void start() throws StartupException {
@Override
public void stop() {
- stopPipeServer();
try {
receiverManager.close();
Review comment:
why delete stop pipe server
--
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] Cpaulyz commented on a change in pull request #5200: [To new_sync][IOTDB-2700] Sync transfer adaption to receiver
Posted by GitBox <gi...@apache.org>.
Cpaulyz commented on a change in pull request #5200:
URL: https://github.com/apache/iotdb/pull/5200#discussion_r825395420
##########
File path: server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java
##########
@@ -0,0 +1,193 @@
+/*
+ * 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.db.newsync.transport;
+
+import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
+import org.apache.iotdb.db.newsync.pipedata.DeletionPipeData;
+import org.apache.iotdb.db.newsync.pipedata.PipeData;
+import org.apache.iotdb.db.newsync.pipedata.SchemaPipeData;
+import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
+import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueue;
+import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueueFactory;
+import org.apache.iotdb.db.newsync.sender.pipe.Pipe;
+import org.apache.iotdb.db.newsync.sender.pipe.TsFilePipe;
+import org.apache.iotdb.db.newsync.transport.client.TransportClient;
+import org.apache.iotdb.db.newsync.transport.server.TransportServerManager;
+import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
+import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import org.apache.commons.io.FileUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class TransportServiceTest {
+ /** create tsfile and move to tmpDir for sync test */
+ File tmpDir = new File("target/synctest");
+
+ String pipeName1 = "pipe1";
+ String remoteIp1 = "127.0.0.1";
+ long createdTime1 = System.currentTimeMillis();
+ File fileDir = new File(SyncPathUtil.getReceiverFileDataDir(pipeName1, remoteIp1, createdTime1));
+ PipeDataQueue pipeDataQueue =
+ PipeDataQueueFactory.getBufferedPipeDataQueue(
+ SyncPathUtil.getReceiverPipeLogDir(pipeName1, remoteIp1, createdTime1));
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.envSetUp();
+ if (!tmpDir.exists()) {
+ tmpDir.mkdirs();
+ }
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ pipeDataQueue.clear();
+ FileUtils.deleteDirectory(tmpDir);
+ EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void test() throws Exception {
+ // 1. prepare fake file
+ File tsfile = new File(tmpDir, "test.tsfile");
+ File resourceFile = new File(tsfile.getAbsoluteFile() + TsFileResource.RESOURCE_SUFFIX);
+ File modsFile = new File(tsfile.getAbsoluteFile() + ModificationFile.FILE_SUFFIX);
+ FileWriter out = new FileWriter(tsfile);
+ out.write("tsfile");
+ out.flush();
+ out.close();
+ out = new FileWriter(resourceFile);
+ out.write("resource");
+ out.flush();
+ out.close();
+ out = new FileWriter(modsFile);
+ out.write("mods");
+ out.flush();
+ out.close();
+
+ // 2. prepare pipelog and pipeDataQueue
+ int serialNum = 0;
+ List<PipeData> pipeDataList = new ArrayList<>();
+ pipeDataList.add(
+ new SchemaPipeData(new SetStorageGroupPlan(new PartialPath("root.vehicle")), serialNum++));
+ pipeDataList.add(
+ new SchemaPipeData(
+ new CreateTimeSeriesPlan(
+ new PartialPath("root.vehicle.d0.s0"),
+ new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.RLE)),
+ serialNum++));
+ TsFilePipeData tsFilePipeData = new TsFilePipeData(tsfile.getPath(), serialNum++);
+ pipeDataList.add(tsFilePipeData);
+ Deletion deletion = new Deletion(new PartialPath("root.vehicle.**"), 0, 33, 38);
+ pipeDataList.add(new DeletionPipeData(deletion, serialNum++));
+
+ // 3. start server
+ TransportServerManager.getInstance().startService();
+
+ // 4. start client
+ Pipe pipe = new TsFilePipe(createdTime1, pipeName1, null, 0, false);
+ TransportClient client = new TransportClient(pipe, "127.0.0.1", 5555);
+ for (PipeData pipeData : pipeDataList) {
+ client.senderTransport(pipeData);
+ }
+
+ // 5. check file
+ Thread.sleep(1000);
+ client.close();
+ TransportServerManager.getInstance().stopService();
+ File[] targetFiles = fileDir.listFiles((dir1, name) -> name.equals(tsfile.getName()));
+ Assert.assertNotNull(targetFiles);
+ Assert.assertEquals(1, targetFiles.length);
+ compareFile(targetFiles[0], tsfile);
+ File[] resourceFiles = fileDir.listFiles((dir1, name) -> name.equals(resourceFile.getName()));
+ Assert.assertNotNull(resourceFiles);
+ Assert.assertEquals(1, resourceFiles.length);
+ compareFile(resourceFiles[0], resourceFile);
+ File[] modsFiles = fileDir.listFiles((dir1, name) -> name.equals(modsFile.getName()));
+ Assert.assertNotNull(modsFiles);
+ Assert.assertEquals(1, modsFiles.length);
+ compareFile(modsFiles[0], modsFile);
+
+ // 6. check pipedata
+ tsFilePipeData.setTsFilePath(fileDir.getAbsolutePath() + File.separator + tsfile.getName());
+ ExecutorService es1 = Executors.newSingleThreadExecutor();
+ List<PipeData> resPipeData = new ArrayList<>();
+ es1.execute(
+ () -> {
+ for (int i = 0; i < pipeDataList.size(); i++) {
+ try {
+ resPipeData.add(pipeDataQueue.take());
+ pipeDataQueue.commit();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ es1.shutdownNow();
+ Assert.assertEquals(pipeDataList.size(), resPipeData.size());
+ for (int i = 0; i < resPipeData.size(); i++) {
+ Assert.assertEquals(pipeDataList.get(i), resPipeData.get(i));
+ }
+ }
+
+ private void compareFile(File firFile, File secFile) {
+ try {
+ BufferedInputStream fir = new BufferedInputStream(new FileInputStream(firFile));
+ BufferedInputStream sec = new BufferedInputStream(new FileInputStream(secFile));
+ // 比较文件的长度是否一样
Review comment:
Yeah, I compare the length and hash of the files now.
--
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] qiaojialin merged pull request #5200: [To new_sync][IOTDB-2700] Sync transfer adaption to receiver
Posted by GitBox <gi...@apache.org>.
qiaojialin merged pull request #5200:
URL: https://github.com/apache/iotdb/pull/5200
--
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] Cpaulyz commented on a change in pull request #5200: [To new_sync][IOTDB-2700] Sync transfer adaption to receiver
Posted by GitBox <gi...@apache.org>.
Cpaulyz commented on a change in pull request #5200:
URL: https://github.com/apache/iotdb/pull/5200#discussion_r825395401
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
##########
@@ -170,31 +182,32 @@ public TransportStatus transportData(
}
if (type != Type.FILE) {
-
buff.position(pos);
int length = buff.capacity();
byte[] byteArray = new byte[length];
buff.get(byteArray);
- try (InputStream inputStream = new ByteArrayInputStream(byteArray);
- DataInputStream dataInputStream = new DataInputStream(inputStream)) {
- PipeData pipeData = PipeData.deserialize(dataInputStream);
- // Do with file
- // BufferedPipeDataQueue.offer(pipeData);
+ try {
+ PipeData pipeData = PipeData.deserialize(byteArray);
+ if (type == Type.TSFILE) {
+ // Do with file
+ handleTsFilePipeData((TsFilePipeData) pipeData, fileDir);
+ }
+ PipeDataQueueFactory.getBufferedPipeDataQueue(getPipeLogDirPath(identityInfo))
+ .offer(pipeData);
} catch (IOException | IllegalPathException e) {
e.printStackTrace();
Review comment:
fixed
--
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] Cpaulyz commented on a change in pull request #5200: [To new_sync][IOTDB-2700] Sync transfer adaption to receiver
Posted by GitBox <gi...@apache.org>.
Cpaulyz commented on a change in pull request #5200:
URL: https://github.com/apache/iotdb/pull/5200#discussion_r825814737
##########
File path: server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
##########
@@ -218,7 +220,6 @@ public void start() throws StartupException {
@Override
public void stop() {
- stopPipeServer();
try {
receiverManager.close();
Review comment:
`stop()` is called when deregister service such as exiting IoTDB.
`stopPipeServer()` stops pipeserver and marks pipeserver unenabled to disk. Receiver service will read disk file and recover status, the correct status of pipeserver should not be unenabled.
--
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