This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch new_sync
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/new_sync by this push:
new 8e630e4 [To new_sync][IOTDB-2700] Sync transfer adaption to receiver
(#5200)
8e630e4 is described below
commit 8e630e45beb5ec55bc465a7b46c61c01bccc4c92
Author: Chen YZ <[email protected]>
AuthorDate: Wed Mar 16 11:01:36 2022 +0800
[To new_sync][IOTDB-2700] Sync transfer adaption to receiver (#5200)
---
.../sync/IoTDBSyncReceiverCollectorIT.java | 10 +-
.../apache/iotdb/db/newsync/conf/SyncPathUtil.java | 5 +
.../iotdb/db/newsync/pipedata/TsFilePipeData.java | 52 +++--
.../pipedata/queue/PipeDataQueueFactory.java | 45 +++++
.../iotdb/db/newsync/receiver/ReceiverService.java | 29 +--
.../db/newsync/receiver/collector/Collector.java | 15 +-
.../db/newsync/receiver/recovery/ReceiverLog.java | 10 +-
.../iotdb/db/newsync/transfer/SyncRequest.java | 64 ------
.../iotdb/db/newsync/transfer/SyncResponse.java | 25 ---
.../newsync/transport/client/ITransportClient.java | 19 ++
.../newsync/transport/client/TransportClient.java | 134 +++++--------
.../db/newsync/transport/conf/TransportConfig.java | 19 ++
.../newsync/transport/conf/TransportConstant.java | 19 ++
.../transport/server/TransportServerManager.java | 31 ++-
.../server/TransportServerManagerMBean.java | 19 ++
.../server/TransportServerThriftHandler.java | 21 +-
.../transport/server/TransportServiceImpl.java | 219 +++++++++++++--------
.../pipedata/BufferedPipeDataQueueTest.java | 12 +-
.../db/newsync/transport/TransportServiceTest.java | 196 ++++++++++++++++++
thrift-sync/src/main/thrift/transport.thrift | 28 ++-
20 files changed, 649 insertions(+), 323 deletions(-)
diff --git
a/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverCollectorIT.java
b/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverCollectorIT.java
index b6f75e9..d64aef7 100644
---
a/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverCollectorIT.java
+++
b/integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverCollectorIT.java
@@ -29,6 +29,7 @@ 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.BufferedPipeDataQueue;
+import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueueFactory;
import org.apache.iotdb.db.newsync.receiver.collector.Collector;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
@@ -189,7 +190,7 @@ public class IoTDBSyncReceiverCollectorIT {
Deletion deletion = new Deletion(new PartialPath("root.vehicle.**"), 0,
33, 38);
PipeData pipeData = new DeletionPipeData(deletion, serialNum++);
BufferedPipeDataQueue pipeDataQueue =
- Collector.getPipeDataQueue(
+ PipeDataQueueFactory.getBufferedPipeDataQueue(
SyncPathUtil.getReceiverPipeLogDir(pipeName1, remoteIp1,
createdTime1));
pipeDataQueue.offer(pipeData);
@@ -275,6 +276,7 @@ public class IoTDBSyncReceiverCollectorIT {
Assert.fail();
}
}
+ pipeDataQueue.clear();
}
@Test
@@ -399,10 +401,10 @@ public class IoTDBSyncReceiverCollectorIT {
// 3. create and start collector
BufferedPipeDataQueue pipeDataQueue1 =
- Collector.getPipeDataQueue(
+ PipeDataQueueFactory.getBufferedPipeDataQueue(
SyncPathUtil.getReceiverPipeLogDir(pipeName1, remoteIp1,
createdTime1));
BufferedPipeDataQueue pipeDataQueue2 =
- Collector.getPipeDataQueue(
+ PipeDataQueueFactory.getBufferedPipeDataQueue(
SyncPathUtil.getReceiverPipeLogDir(pipeName2, remoteIp2,
createdTime2));
Collector collector = new Collector();
collector.startCollect();
@@ -499,5 +501,7 @@ public class IoTDBSyncReceiverCollectorIT {
Assert.fail();
}
}
+ pipeDataQueue2.clear();
+ pipeDataQueue1.clear();
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncPathUtil.java
b/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncPathUtil.java
index 50f024e..143a7f9 100644
--- a/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncPathUtil.java
+++ b/server/src/main/java/org/apache/iotdb/db/newsync/conf/SyncPathUtil.java
@@ -24,6 +24,11 @@ import java.io.File;
/** Util for path generation in sync module */
public class SyncPathUtil {
+
+ private SyncPathUtil() {
+ // forbidding instantiation
+ }
+
/** sender */
public static String getSenderPipeDir(String pipeName, long createTime) {
return IoTDBDescriptor.getInstance().getConfig().getNewSyncDir()
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/TsFilePipeData.java
b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/TsFilePipeData.java
index aec255d..115af1a 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/TsFilePipeData.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/TsFilePipeData.java
@@ -41,15 +41,33 @@ import java.util.Objects;
public class TsFilePipeData extends PipeData {
private static final Logger logger =
LoggerFactory.getLogger(TsFilePipeData.class);
- private String tsFilePath;
+ private String parentDirPath;
+ private String tsFileName;
public TsFilePipeData(String tsFilePath, long serialNumber) {
super(serialNumber);
- this.tsFilePath = tsFilePath;
+ String sep = File.separator.equals("\\") ? "\\\\" : File.separator;
+ String[] paths = tsFilePath.split(sep);
+ tsFileName = paths[paths.length - 1];
+ parentDirPath = tsFilePath.substring(0, tsFilePath.length() -
tsFileName.length());
}
- public void setTsFilePath(String tsFilePath) {
- this.tsFilePath = tsFilePath;
+ public TsFilePipeData(String parentDirPath, String tsFileName, long
serialNumber) {
+ super(serialNumber);
+ this.parentDirPath = parentDirPath;
+ this.tsFileName = tsFileName;
+ }
+
+ public void setParentDirPath(String parentDirPath) {
+ this.parentDirPath = parentDirPath;
+ }
+
+ public String getTsFileName() {
+ return tsFileName;
+ }
+
+ public String getTsFilePath() {
+ return parentDirPath + File.separator + tsFileName;
}
@Override
@@ -59,18 +77,21 @@ public class TsFilePipeData extends PipeData {
@Override
public long serialize(DataOutputStream stream) throws IOException {
- return super.serialize(stream) + ReadWriteIOUtils.write(tsFilePath,
stream);
+ return super.serialize(stream)
+ + ReadWriteIOUtils.write(parentDirPath, stream)
+ + ReadWriteIOUtils.write(tsFileName, stream);
}
public static TsFilePipeData deserialize(DataInputStream stream) throws
IOException {
long serialNumber = stream.readLong();
- String tsFilePath = ReadWriteIOUtils.readString(stream);
- return new TsFilePipeData(tsFilePath, serialNumber);
+ String parentDirPath = ReadWriteIOUtils.readString(stream);
+ String tsFileName = ReadWriteIOUtils.readString(stream);
+ return new TsFilePipeData(parentDirPath == null ? "" : parentDirPath,
tsFileName, serialNumber);
}
@Override
public ILoader createLoader() {
- return new TsFileLoader(new File(tsFilePath));
+ return new TsFileLoader(new File(getTsFilePath()));
}
@Override
@@ -82,7 +103,7 @@ public class TsFilePipeData extends PipeData {
}
public List<File> getTsFiles() throws FileNotFoundException {
- File tsFile = new File(tsFilePath).getAbsoluteFile();
+ File tsFile = new File(getTsFilePath()).getAbsoluteFile();
File resource = new File(tsFile.getAbsolutePath() +
TsFileResource.RESOURCE_SUFFIX);
File mods = new File(tsFile.getAbsolutePath() +
ModificationFile.FILE_SUFFIX);
@@ -108,18 +129,18 @@ public class TsFilePipeData extends PipeData {
try {
Thread.sleep(SyncConstant.DEFAULT_WAITING_FOR_TSFILE_CLOSE_MILLISECONDS);
} catch (InterruptedException e) {
- logger.warn(String.format("Be Interrupted when waiting for tsfile %s
closed", tsFilePath));
+ logger.warn(String.format("Be Interrupted when waiting for tsfile %s
closed", tsFileName));
}
logger.info(
String.format(
"Waiting for tsfile %s close, retry %d / %d.",
- tsFilePath, (i + 1),
SyncConstant.DEFAULT_WAITING_FOR_TSFILE_RETRY_NUMBER));
+ tsFileName, (i + 1),
SyncConstant.DEFAULT_WAITING_FOR_TSFILE_RETRY_NUMBER));
}
return false;
}
private boolean isTsFileClosed() {
- File tsFile = new File(tsFilePath).getAbsoluteFile();
+ File tsFile = new File(getTsFilePath()).getAbsoluteFile();
File resource = new File(tsFile.getAbsolutePath() +
TsFileResource.RESOURCE_SUFFIX);
return resource.exists();
}
@@ -130,7 +151,7 @@ public class TsFilePipeData extends PipeData {
+ "serialNumber="
+ serialNumber
+ ", tsFilePath='"
- + tsFilePath
+ + getTsFilePath()
+ '\''
+ '}';
}
@@ -140,12 +161,13 @@ public class TsFilePipeData extends PipeData {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
TsFilePipeData pipeData = (TsFilePipeData) o;
- return Objects.equals(tsFilePath, pipeData.tsFilePath)
+ return Objects.equals(parentDirPath, pipeData.parentDirPath)
+ && Objects.equals(tsFileName, pipeData.tsFileName)
&& Objects.equals(serialNumber, pipeData.serialNumber);
}
@Override
public int hashCode() {
- return Objects.hash(tsFilePath, serialNumber);
+ return Objects.hash(parentDirPath, tsFileName, serialNumber);
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/PipeDataQueueFactory.java
b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/PipeDataQueueFactory.java
new file mode 100644
index 0000000..92d1f70
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/pipedata/queue/PipeDataQueueFactory.java
@@ -0,0 +1,45 @@
+/*
+ * 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.pipedata.queue;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class PipeDataQueueFactory {
+
+ // TODO: try use weakReference to avoid memory leak
+ private static final Map<String, BufferedPipeDataQueue>
bufferedPipeDataQueueMap =
+ new ConcurrentHashMap<>();
+ /**
+ * get or create BufferedPipeDataQueue identified by key
+ *
+ * @param pipeLogDir using path of pipe-log dir as key
+ * @return BufferedPipeDataQueue
+ */
+ public static BufferedPipeDataQueue getBufferedPipeDataQueue(String
pipeLogDir) {
+ return bufferedPipeDataQueueMap.computeIfAbsent(
+ pipeLogDir, i -> new BufferedPipeDataQueue(pipeLogDir));
+ }
+
+ public static void removeBufferedPipeDataQueue(String pipeLogDir) {
+ BufferedPipeDataQueue queue = bufferedPipeDataQueueMap.remove(pipeLogDir);
+ queue.clear();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
b/server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
index bfdb672..983f17c 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/receiver/ReceiverService.java
@@ -21,18 +21,20 @@ package org.apache.iotdb.db.newsync.receiver;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
+import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueueFactory;
import org.apache.iotdb.db.newsync.receiver.collector.Collector;
import org.apache.iotdb.db.newsync.receiver.manager.PipeInfo;
import org.apache.iotdb.db.newsync.receiver.manager.PipeMessage;
import org.apache.iotdb.db.newsync.receiver.manager.PipeStatus;
import org.apache.iotdb.db.newsync.receiver.manager.ReceiverManager;
-import org.apache.iotdb.db.newsync.transfer.SyncRequest;
-import org.apache.iotdb.db.newsync.transfer.SyncResponse;
+import org.apache.iotdb.db.newsync.transport.server.TransportServerManager;
import org.apache.iotdb.db.qp.physical.sys.ShowPipeServerPlan;
import org.apache.iotdb.db.qp.utils.DatetimeUtils;
import org.apache.iotdb.db.query.dataset.ListDataSet;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.service.transport.thrift.SyncRequest;
+import org.apache.iotdb.service.transport.thrift.SyncResponse;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
@@ -69,8 +71,8 @@ public class ReceiverService implements IService {
pipeInfo.getPipeName(), pipeInfo.getRemoteIp(),
pipeInfo.getCreateTime());
}
}
- // TODO: start socket
- } catch (IOException e) {
+ TransportServerManager.getInstance().startService();
+ } catch (IOException | StartupException e) {
logger.error(e.getMessage());
return false;
}
@@ -82,7 +84,8 @@ public class ReceiverService implements IService {
try {
receiverManager.stopServer();
collector.stopCollect();
- // TODO: stop socket and collector
+ // todo: how to stop?
+ TransportServerManager.getInstance().stopService();
} catch (IOException e) {
logger.error(e.getMessage());
return false;
@@ -94,22 +97,22 @@ public class ReceiverService implements IService {
// TODO: define exception
// TODO: this is a mock interface
public SyncResponse recMsg(SyncRequest request) throws IOException {
- switch (request.getCode()) {
- case SyncRequest.HEARTBEAT:
+ switch (request.getType()) {
+ case HEARTBEAT:
List<PipeMessage> messages =
receiverManager.getPipeMessages(
request.getPipeName(), request.getRemoteIp(),
request.getCreateTime());
break;
- case SyncRequest.CREATE:
+ case CREATE:
createPipe(request.getPipeName(), request.getRemoteIp(),
request.getCreateTime());
break;
- case SyncRequest.START:
+ case START:
startPipe(request.getPipeName(), request.getRemoteIp(),
request.getCreateTime());
break;
- case SyncRequest.STOP:
+ case STOP:
stopPipe(request.getPipeName(), request.getRemoteIp(),
request.getCreateTime());
break;
- case SyncRequest.DROP:
+ case DROP:
dropPipe(request.getPipeName(), request.getRemoteIp(),
request.getCreateTime());
break;
}
@@ -142,6 +145,8 @@ public class ReceiverService implements IService {
collector.stopPipe(pipeName, remoteIp, createTime);
File dir = new File(SyncPathUtil.getReceiverPipeDir(pipeName, remoteIp,
createTime));
FileUtils.deleteDirectory(dir);
+ PipeDataQueueFactory.removeBufferedPipeDataQueue(
+ SyncPathUtil.getReceiverPipeLogDir(pipeName, remoteIp, createTime));
}
private void createDir(String pipeName, String remoteIp, long createTime) {
@@ -218,9 +223,9 @@ public class ReceiverService implements IService {
@Override
public void stop() {
- stopPipeServer();
try {
receiverManager.close();
+ collector.stopCollect();
} catch (IOException e) {
logger.error(e.getMessage());
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/receiver/collector/Collector.java
b/server/src/main/java/org/apache/iotdb/db/newsync/receiver/collector/Collector.java
index 6d85631..3c5b309 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/receiver/collector/Collector.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/receiver/collector/Collector.java
@@ -24,11 +24,10 @@ import org.apache.iotdb.db.concurrent.ThreadName;
import org.apache.iotdb.db.exception.metadata.StorageGroupAlreadySetException;
import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
import org.apache.iotdb.db.newsync.pipedata.PipeData;
-import org.apache.iotdb.db.newsync.pipedata.queue.BufferedPipeDataQueue;
import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueue;
+import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueueFactory;
import org.apache.iotdb.db.newsync.receiver.manager.PipeMessage;
import org.apache.iotdb.db.newsync.receiver.manager.ReceiverManager;
-import org.apache.iotdb.db.utils.TestOnly;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,15 +50,6 @@ public class Collector {
taskFutures = new ConcurrentHashMap<>();
}
- private static Map<String, BufferedPipeDataQueue> bufferedPipeDataQueueMap =
- new ConcurrentHashMap<>();
-
- @TestOnly
- public static BufferedPipeDataQueue getPipeDataQueue(String pipeLogDir) {
- return bufferedPipeDataQueueMap.computeIfAbsent(
- pipeLogDir, i -> new BufferedPipeDataQueue(pipeLogDir));
- }
-
public void startCollect() {
this.executorService =
IoTDBThreadPoolFactory.newCachedThreadPool(ThreadName.SYNC_RECEIVER_COLLECTOR.getName());
@@ -118,7 +108,8 @@ public class Collector {
@Override
public void run() {
PipeDataQueue pipeDataQueue =
- getPipeDataQueue(SyncPathUtil.getReceiverPipeLogDir(pipeName,
remoteIp, createTime));
+ PipeDataQueueFactory.getBufferedPipeDataQueue(
+ SyncPathUtil.getReceiverPipeLogDir(pipeName, remoteIp,
createTime));
while (!Thread.interrupted()) {
PipeData pipeData = null;
try {
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/receiver/recovery/ReceiverLog.java
b/server/src/main/java/org/apache/iotdb/db/newsync/receiver/recovery/ReceiverLog.java
index abbc710..03ec25a 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/receiver/recovery/ReceiverLog.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/receiver/recovery/ReceiverLog.java
@@ -115,7 +115,13 @@ public class ReceiverLog {
}
public void close() throws IOException {
- bw.close();
- msg.close();
+ if (bw != null) {
+ bw.close();
+ bw = null;
+ }
+ if (msg != null) {
+ msg.close();
+ msg = null;
+ }
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transfer/SyncRequest.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transfer/SyncRequest.java
deleted file mode 100644
index c49355c..0000000
--- a/server/src/main/java/org/apache/iotdb/db/newsync/transfer/SyncRequest.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.transfer;
-
-public class SyncRequest {
- public static final int CREATE = 0;
- public static final int START = 1;
- public static final int STOP = 2;
- public static final int DROP = 3;
- public static final int HEARTBEAT = 4;
-
- private int code;
- private String pipeName;
- private String remoteIp;
- private long createTime;
-
- public int getCode() {
- return code;
- }
-
- public void setCode(int code) {
- this.code = code;
- }
-
- public String getPipeName() {
- return pipeName;
- }
-
- public void setPipeName(String pipeName) {
- this.pipeName = pipeName;
- }
-
- public String getRemoteIp() {
- return remoteIp;
- }
-
- public void setRemoteIp(String remoteIp) {
- this.remoteIp = remoteIp;
- }
-
- public long getCreateTime() {
- return createTime;
- }
-
- public void setCreateTime(long createTime) {
- this.createTime = createTime;
- }
-}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transfer/SyncResponse.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transfer/SyncResponse.java
deleted file mode 100644
index 5f5eb1e..0000000
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transfer/SyncResponse.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.transfer;
-
-public class SyncResponse {
- public static final int SUCCESS = 0;
- public static final int WARN = 1;
- public static final int ERROR = 2;
-}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/ITransportClient.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/ITransportClient.java
index 0499fb1..1376dcc 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/ITransportClient.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/ITransportClient.java
@@ -1,3 +1,22 @@
+/*
+ * 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.client;
public interface ITransportClient {}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/TransportClient.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/TransportClient.java
index 037543b..0179656 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/TransportClient.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/client/TransportClient.java
@@ -1,3 +1,22 @@
+/*
+ * 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.client;
import org.apache.iotdb.db.concurrent.ThreadName;
@@ -8,10 +27,8 @@ import org.apache.iotdb.db.newsync.pipedata.PipeData;
import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
import org.apache.iotdb.db.newsync.sender.pipe.Pipe;
import org.apache.iotdb.db.newsync.transport.conf.TransportConstant;
-import org.apache.iotdb.db.sync.conf.SyncConstant;
import org.apache.iotdb.db.sync.conf.SyncSenderConfig;
import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
-import org.apache.iotdb.db.sync.sender.transfer.SyncClient;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.rpc.RpcTransportFactory;
import org.apache.iotdb.service.transport.thrift.IdentityInfo;
@@ -31,13 +48,8 @@ import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.BufferedReader;
-import java.io.ByteArrayOutputStream;
-import java.io.DataOutputStream;
import java.io.File;
import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
@@ -45,7 +57,6 @@ import java.net.Socket;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
-import java.util.UUID;
import static
org.apache.iotdb.db.newsync.transport.conf.TransportConfig.isCheckFileDegistAgain;
import static
org.apache.iotdb.db.newsync.transport.conf.TransportConstant.REBASE_CODE;
@@ -54,7 +65,7 @@ import static
org.apache.iotdb.db.newsync.transport.conf.TransportConstant.SUCCE
public class TransportClient implements ITransportClient, Runnable {
- private static final Logger logger =
LoggerFactory.getLogger(SyncClient.class);
+ private static final Logger logger =
LoggerFactory.getLogger(TransportClient.class);
// TODO: Need to change to transport config
private static SyncSenderConfig config =
SyncSenderDescriptor.getInstance().getConfig();
@@ -71,8 +82,6 @@ public class TransportClient implements ITransportClient,
Runnable {
private int port = -1;
- private String uuid = null;
-
private IdentityInfo identityInfo = null;
private Pipe pipe = null;
@@ -91,7 +100,6 @@ public class TransportClient implements ITransportClient,
Runnable {
public void setServerConfig(String ipAddress, int port) throws IOException {
this.ipAddress = ipAddress;
this.port = port;
- this.uuid = getOrCreateUUID(getUuidFile());
}
public TransportClient(Pipe pipe, String ipAddress, int port) throws
IOException {
@@ -101,7 +109,6 @@ public class TransportClient implements ITransportClient,
Runnable {
this.pipe = pipe;
this.ipAddress = ipAddress;
this.port = port;
- this.uuid = getOrCreateUUID(getUuidFile());
handshake();
}
@@ -146,7 +153,8 @@ public class TransportClient implements ITransportClient,
Runnable {
identityInfo =
new IdentityInfo(
socket.getLocalAddress().getHostAddress(),
- this.uuid,
+ pipe.getName(),
+ pipe.getCreateTime(),
ioTDBConfig.getIoTDBMajorVersion());
TransportStatus status = serviceClient.handshake(identityInfo);
if (status.code != SUCCESS_CODE) {
@@ -165,7 +173,7 @@ public class TransportClient implements ITransportClient,
Runnable {
return true;
}
- private boolean senderTransport(PipeData pipeData) {
+ public boolean senderTransport(PipeData pipeData) {
int retryCount = 0;
@@ -272,17 +280,15 @@ public class TransportClient implements ITransportClient,
Runnable {
}
int dataLength;
- try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r");
- ByteArrayOutputStream byteArrayOutputStream =
- new ByteArrayOutputStream(TransportConstant.DATA_CHUNK_SIZE)) {
-
+ try (RandomAccessFile randomAccessFile = new RandomAccessFile(file,
"r")) {
+ if (randomAccessFile.length() <= position) {
+ break;
+ }
randomAccessFile.seek(position);
while ((dataLength = randomAccessFile.read(buffer)) != -1) {
messageDigest.reset();
- byteArrayOutputStream.write(buffer, 0, dataLength);
messageDigest.update(buffer, 0, dataLength);
- ByteBuffer buffToSend =
ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
- byteArrayOutputStream.reset();
+ ByteBuffer buffToSend = ByteBuffer.wrap(buffer, 0, dataLength);
MetaInfo metaInfo = new MetaInfo(Type.FILE, file.getName(),
position);
TransportStatus status = null;
@@ -351,16 +357,11 @@ public class TransportClient implements ITransportClient,
Runnable {
"Can not sync pipe data after %s tries.",
config.getMaxNumOfSyncFileRetry()));
}
- try (ByteArrayOutputStream byteArrayOutputStream = new
ByteArrayOutputStream();
- DataOutputStream dataOutputStream = new
DataOutputStream(byteArrayOutputStream)) {
- int dataLength = new
Long(pipeData.serialize(dataOutputStream)).intValue();
- byte[] buffer = new byte[dataLength];
-
- byteArrayOutputStream.write(buffer, 0, dataLength);
+ try {
+ byte[] buffer = pipeData.serialize();
messageDigest.reset();
- messageDigest.update(buffer, 0, dataLength);
- ByteBuffer buffToSend =
ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
- byteArrayOutputStream.reset();
+ messageDigest.update(buffer);
+ ByteBuffer buffToSend = ByteBuffer.wrap(buffer);
MetaInfo metaInfo =
new MetaInfo(Type.findByValue(pipeData.getType().ordinal()),
"fileName", 0);
@@ -380,49 +381,6 @@ public class TransportClient implements ITransportClient,
Runnable {
}
}
- /** UUID marks the identity of sender for receiver. */
- private String getOrCreateUUID(File uuidFile) throws IOException {
- if (!uuidFile.getParentFile().exists()) {
- uuidFile.getParentFile().mkdirs();
- }
-
- String uuid;
- if (uuidFile.exists()) {
- try (BufferedReader bf = new BufferedReader((new FileReader(uuidFile))))
{
- uuid = bf.readLine();
- } catch (IOException e) {
- logger.error("Cannot read UUID from file {}", uuidFile.getPath());
- throw new IOException(e);
- }
-
- if ((uuid == null) || (uuid.length() == 0)) {
- logger.warn("UUID in file {} is empty.", uuidFile.getPath());
- uuidFile.delete();
- } else {
- return uuid;
- }
- }
-
- // uuidFile not exist or uuid in uuidFile is invalid
- try (FileOutputStream out = new FileOutputStream(uuidFile)) {
- uuid = generateUUID();
- out.write(uuid.getBytes());
- } catch (IOException e) {
- logger.error("Cannot insert UUID to file {}", uuidFile.getPath());
- throw new IOException(e);
- }
-
- return uuid;
- }
-
- private String generateUUID() {
- return UUID.randomUUID().toString().replaceAll("-", "");
- }
-
- private File getUuidFile() {
- return new File(ioTDBConfig.getSyncDir(), SyncConstant.UUID_FILE_NAME);
- }
-
/**
* When an object implementing interface <code>Runnable</code> is used to
create a thread,
* starting the thread causes the object's <code>run</code> method to be
called in that separately
@@ -471,21 +429,25 @@ public class TransportClient implements ITransportClient,
Runnable {
return;
}
- // Example 1. Send TSFILE.
- // List<File> files = new ArrayList<>();
- // files.add(new File(System.getProperty(IoTDBConstant.IOTDB_HOME) +
"/files/test1"));
- // files.add(new File(System.getProperty(IoTDBConstant.IOTDB_HOME) +
"/files/test2"));
- // files.add(new File(System.getProperty(IoTDBConstant.IOTDB_HOME) +
"/files/test3"));
+ // Example 1. Send TSFILE.
+ // List<File> files = new ArrayList<>();
+ // files.add(new File(System.getProperty(IoTDBConstant.IOTDB_HOME)
+ "/files/test1"));
+ // files.add(new File(System.getProperty(IoTDBConstant.IOTDB_HOME)
+ "/files/test2"));
+ // files.add(new File(System.getProperty(IoTDBConstant.IOTDB_HOME)
+ "/files/test3"));
+
+ // if (!TransportClient.getInstance().senderTransport()) {
+ // Deal with the error here.
+ // }
- // if (!TransportClient.getInstance().senderTransport()) {
- // Deal with the error here.
- // }
+ // Example 2. Send DELETION
+ // TsFilePipeData.Type.DELETION.name();
- // Example 2. Send DELETION
- // TsFilePipeData.Type.DELETION.name();
+ // Example 3. Send PHYSICALPLAN
+ // TsFilePipeData.Type.PHYSICALPLAN.name();
+ }
- // Example 3. Send PHYSICALPLAN
- // TsFilePipeData.Type.PHYSICALPLAN.name();
+ public void close() {
+ transport.close();
}
private static class InstanceHolder {
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportConfig.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportConfig.java
index 316f6f5..e9af083 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportConfig.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportConfig.java
@@ -1,3 +1,22 @@
+/*
+ * 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.conf;
import org.apache.iotdb.db.conf.IoTDBConstant;
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportConstant.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportConstant.java
index daf1d0d..7d575c7 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportConstant.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/conf/TransportConstant.java
@@ -1,3 +1,22 @@
+/*
+ * 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.conf;
import org.apache.iotdb.rpc.RpcUtils;
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManager.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManager.java
index 7f9a540..c9d8c64 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManager.java
@@ -1,3 +1,22 @@
+/*
+ * 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.server;
import org.apache.iotdb.db.concurrent.ThreadName;
@@ -92,18 +111,18 @@ public class TransportServerManager extends ThriftService
@Override
public void startService() throws StartupException {
// TODO: Whether to change this config here
- if (!IoTDBDescriptor.getInstance().getConfig().isSyncEnable()) {
- return;
- }
+ // if (!IoTDBDescriptor.getInstance().getConfig().isSyncEnable()) {
+ // return;
+ // }
super.startService();
}
@Override
public void stopService() {
// TODO: Whether to change this config here
- if (IoTDBDescriptor.getInstance().getConfig().isSyncEnable()) {
- super.stopService();
- }
+ // if (IoTDBDescriptor.getInstance().getConfig().isSyncEnable()) {
+ super.stopService();
+ // }
}
@TestOnly
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManagerMBean.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManagerMBean.java
index fec1947..ebb892b 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManagerMBean.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerManagerMBean.java
@@ -1,3 +1,22 @@
+/*
+ * 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.server;
import org.apache.iotdb.db.exception.StartupException;
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerThriftHandler.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerThriftHandler.java
index 8a3898c..5815317 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerThriftHandler.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServerThriftHandler.java
@@ -1,3 +1,22 @@
+/*
+ * 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.server;
import org.apache.thrift.protocol.TProtocol;
@@ -9,7 +28,7 @@ public class TransportServerThriftHandler implements
TServerEventHandler {
private TransportServiceImpl serviceImpl;
- TransportServerThriftHandler(TransportServiceImpl serviceImpl) {
+ public TransportServerThriftHandler(TransportServiceImpl serviceImpl) {
this.serviceImpl = serviceImpl;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
index 0160cf3..cdfa9db 100644
---
a/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
+++
b/server/src/main/java/org/apache/iotdb/db/newsync/transport/server/TransportServiceImpl.java
@@ -1,9 +1,31 @@
+/*
+ * 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.server;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.newsync.conf.SyncPathUtil;
import org.apache.iotdb.db.newsync.pipedata.PipeData;
+import org.apache.iotdb.db.newsync.pipedata.TsFilePipeData;
+import org.apache.iotdb.db.newsync.pipedata.queue.PipeDataQueueFactory;
import org.apache.iotdb.service.transport.thrift.IdentityInfo;
import org.apache.iotdb.service.transport.thrift.MetaInfo;
import org.apache.iotdb.service.transport.thrift.SyncRequest;
@@ -16,16 +38,7 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.BufferedReader;
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.io.RandomAccessFile;
+import java.io.*;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.file.Files;
@@ -34,7 +47,6 @@ import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
-import static
org.apache.iotdb.db.newsync.transport.conf.TransportConfig.getSyncedDir;
import static
org.apache.iotdb.db.newsync.transport.conf.TransportConstant.CONFLICT_CODE;
import static
org.apache.iotdb.db.newsync.transport.conf.TransportConstant.ERROR_CODE;
import static
org.apache.iotdb.db.newsync.transport.conf.TransportConstant.REBASE_CODE;
@@ -46,6 +58,8 @@ public class TransportServiceImpl implements
TransportService.Iface {
private static Logger logger =
LoggerFactory.getLogger(TransportServiceImpl.class);
private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private static final String RECORD_SUFFIX = ".record";
+ private static final String PATCH_SUFFIX = ".patch";
private class CheckResult {
boolean result;
@@ -66,7 +80,7 @@ public class TransportServiceImpl implements
TransportService.Iface {
}
private CheckResult checkStartIndexValid(File file, long startIndex) throws
IOException {
- File recordFile = new File(file.getAbsolutePath() + ".record");
+ File recordFile = new File(file.getAbsolutePath() + RECORD_SUFFIX);
if (!recordFile.exists() && startIndex != 0) {
logger.error(
@@ -121,8 +135,8 @@ public class TransportServiceImpl implements
TransportService.Iface {
identityInfo.version, config.getIoTDBVersion()));
}
- if (!new File(getSyncedDir(identityInfo.getAddress(),
identityInfo.getUuid())).exists()) {
- new File(getSyncedDir(identityInfo.getAddress(),
identityInfo.getUuid())).mkdirs();
+ if (!new File(getFileDataDirPath(identityInfo)).exists()) {
+ new File(getFileDataDirPath(identityInfo)).mkdirs();
}
return new TransportStatus(SUCCESS_CODE, "");
}
@@ -132,82 +146,79 @@ public class TransportServiceImpl implements
TransportService.Iface {
IdentityInfo identityInfo, MetaInfo metaInfo, ByteBuffer buff,
ByteBuffer digest) {
logger.debug("Invoke transportData method from client ip = {}",
identityInfo.address);
- String ipAddress = identityInfo.address;
- String uuid = identityInfo.uuid;
- synchronized (uuid.intern()) {
- Type type = metaInfo.type;
- String fileName = metaInfo.fileName;
- long startIndex = metaInfo.startIndex;
-
- // Check file start index valid
- if (type == Type.FILE) {
- try {
- CheckResult result =
- checkStartIndexValid(new File(getSyncedDir(ipAddress, uuid),
fileName), startIndex);
- if (!result.isResult()) {
- return new TransportStatus(REBASE_CODE, result.getIndex());
- }
- } catch (IOException e) {
- logger.error(e.getMessage());
- return new TransportStatus(ERROR_CODE, e.getMessage());
- }
- }
+ String fileDir = getFileDataDirPath(identityInfo);
+ Type type = metaInfo.type;
+ String fileName = metaInfo.fileName;
+ long startIndex = metaInfo.startIndex;
- // Check buff digest
- int pos = buff.position();
- MessageDigest messageDigest = null;
+ // Check file start index valid
+ if (type == Type.FILE) {
try {
- messageDigest = MessageDigest.getInstance("SHA-256");
- } catch (NoSuchAlgorithmException e) {
+ CheckResult result = checkStartIndexValid(new File(fileDir, fileName),
startIndex);
+ if (!result.isResult()) {
+ return new TransportStatus(REBASE_CODE, result.getIndex());
+ }
+ } catch (IOException e) {
logger.error(e.getMessage());
return new TransportStatus(ERROR_CODE, e.getMessage());
}
- messageDigest.update(buff);
- byte[] digestBytes = new byte[digest.capacity()];
- digest.get(digestBytes);
- if (!Arrays.equals(messageDigest.digest(), digestBytes)) {
- return new TransportStatus(RETRY_CODE, "Data digest check error,
retry.");
- }
+ }
- if (type != Type.FILE) {
+ // Check buff digest
+ int pos = buff.position();
+ MessageDigest messageDigest = null;
+ try {
+ messageDigest = MessageDigest.getInstance("SHA-256");
+ } catch (NoSuchAlgorithmException e) {
+ logger.error(e.getMessage());
+ return new TransportStatus(ERROR_CODE, e.getMessage());
+ }
+ messageDigest.update(buff);
+ byte[] digestBytes = new byte[digest.capacity()];
+ digest.get(digestBytes);
+ if (!Arrays.equals(messageDigest.digest(), digestBytes)) {
+ return new TransportStatus(RETRY_CODE, "Data digest check error,
retry.");
+ }
- buff.position(pos);
+ if (type != Type.FILE) {
+ buff.position(pos);
+ int length = buff.capacity();
+ byte[] byteArray = new byte[length];
+ buff.get(byteArray);
+ 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) {
+ logger.error("Pipe data transport error, {}", e.getMessage());
+ return new TransportStatus(RETRY_CODE, "Data digest transport error "
+ e.getMessage());
+ }
+ } else {
+ // Write buff to {file}.patch
+ buff.position(pos);
+ File file = new File(fileDir, fileName + PATCH_SUFFIX);
+ try (RandomAccessFile randomAccessFile = new RandomAccessFile(file,
"rw")) {
+ randomAccessFile.seek(startIndex);
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);
- } catch (IOException | IllegalPathException e) {
- e.printStackTrace();
- }
- } else {
- // Write buff to {file}.patch
- buff.position(pos);
- File file = new File(getSyncedDir(ipAddress, uuid), fileName +
".patch");
- try (RandomAccessFile randomAccessFile = new RandomAccessFile(file,
"rw")) {
- randomAccessFile.seek(startIndex);
- int length = buff.capacity();
- byte[] byteArray = new byte[length];
- buff.get(byteArray);
- randomAccessFile.write(byteArray);
- writeRecordFile(
- new File(getSyncedDir(ipAddress, uuid), fileName + ".record"),
startIndex + length);
- logger.debug(
- "Sync "
- + fileName
- + " start at "
- + startIndex
- + " to "
- + (startIndex + length)
- + " is done.");
- } catch (IOException e) {
- logger.error(e.getMessage());
- e.printStackTrace();
- return new TransportStatus(ERROR_CODE, e.getMessage());
- }
+ randomAccessFile.write(byteArray);
+ writeRecordFile(new File(fileDir, fileName + RECORD_SUFFIX),
startIndex + length);
+ logger.debug(
+ "Sync "
+ + fileName
+ + " start at "
+ + startIndex
+ + " to "
+ + (startIndex + length)
+ + " is done.");
+ } catch (IOException e) {
+ logger.error(e.getMessage());
+ return new TransportStatus(ERROR_CODE, e.getMessage());
}
}
return new TransportStatus(SUCCESS_CODE, "");
@@ -218,9 +229,8 @@ public class TransportServiceImpl implements
TransportService.Iface {
IdentityInfo identityInfo, MetaInfo metaInfo, ByteBuffer digest) throws
TException {
logger.debug("Invoke checkFileDigest method from client ip = {}",
identityInfo.address);
- String ipAddress = identityInfo.getAddress();
- String uuid = identityInfo.getUuid();
- synchronized (uuid.intern()) {
+ String fileDir = getFileDataDirPath(identityInfo);
+ synchronized (fileDir.intern()) {
String fileName = metaInfo.fileName;
MessageDigest messageDigest = null;
try {
@@ -231,7 +241,7 @@ public class TransportServiceImpl implements
TransportService.Iface {
}
try (InputStream inputStream =
- new FileInputStream(new File(getSyncedDir(ipAddress, uuid), fileName
+ ".patch"))) {
+ new FileInputStream(new File(fileDir, fileName + PATCH_SUFFIX))) {
byte[] block = new byte[DATA_CHUNK_SIZE];
int length;
while ((length = inputStream.read(block)) > 0) {
@@ -248,11 +258,11 @@ public class TransportServiceImpl implements
TransportService.Iface {
fileName,
localDigest,
digest);
- new File(getSyncedDir(ipAddress, uuid), fileName +
".record").delete();
+ new File(fileDir, fileName + RECORD_SUFFIX).delete();
return new TransportStatus(CONFLICT_CODE, "File digest check
error.");
}
} catch (IOException e) {
- e.printStackTrace();
+ logger.error(e.getMessage());
return new TransportStatus(ERROR_CODE, e.getMessage());
}
@@ -283,4 +293,41 @@ public class TransportServiceImpl implements
TransportService.Iface {
// TODO: Handle client exit here.
// do nothing now
}
+
+ /**
+ * handle when successfully receive tsFilePipeData. Rename .patch file and
reset tsFilePipeData's
+ * path.
+ *
+ * @param tsFilePipeData pipeData
+ * @param fileDir path of file data dir
+ */
+ private void handleTsFilePipeData(TsFilePipeData tsFilePipeData, String
fileDir) {
+ String tsFileName = tsFilePipeData.getTsFileName();
+ File dir = new File(fileDir);
+ File[] targetFiles =
+ dir.listFiles((dir1, name) -> name.startsWith(tsFileName) &&
name.endsWith(PATCH_SUFFIX));
+ // TODO: same name ?
+ if (targetFiles != null) {
+ for (File targetFile : targetFiles) {
+ File newFile =
+ new File(
+ dir,
+ targetFile
+ .getName()
+ .substring(0, targetFile.getName().length() -
PATCH_SUFFIX.length()));
+ targetFile.renameTo(newFile);
+ }
+ }
+ tsFilePipeData.setParentDirPath(dir.getAbsolutePath());
+ }
+
+ private String getFileDataDirPath(IdentityInfo identityInfo) {
+ return SyncPathUtil.getReceiverFileDataDir(
+ identityInfo.getPipeName(), identityInfo.getAddress(),
identityInfo.getCreateTime());
+ }
+
+ private String getPipeLogDirPath(IdentityInfo identityInfo) {
+ return SyncPathUtil.getReceiverPipeLogDir(
+ identityInfo.getPipeName(), identityInfo.getAddress(),
identityInfo.getCreateTime());
+ }
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/newsync/pipedata/BufferedPipeDataQueueTest.java
b/server/src/test/java/org/apache/iotdb/db/newsync/pipedata/BufferedPipeDataQueueTest.java
index 33afae8..e187cce 100644
---
a/server/src/test/java/org/apache/iotdb/db/newsync/pipedata/BufferedPipeDataQueueTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/newsync/pipedata/BufferedPipeDataQueueTest.java
@@ -74,7 +74,7 @@ public class BufferedPipeDataQueueTest {
new FileOutputStream(
new File(pipeLogDir.getPath(),
SyncConstant.getPipeLogName(0)), false));
for (int i = 0; i < 4; i++) {
- new TsFilePipeData(null, i).serialize(pipeLogOutput1);
+ new TsFilePipeData("", i).serialize(pipeLogOutput1);
}
pipeLogOutput1.close();
// pipelog2: 4~10
@@ -83,7 +83,7 @@ public class BufferedPipeDataQueueTest {
new FileOutputStream(
new File(pipeLogDir.getPath(),
SyncConstant.getPipeLogName(4)), false));
for (int i = 4; i < 11; i++) {
- new TsFilePipeData(null, i).serialize(pipeLogOutput2);
+ new TsFilePipeData("", i).serialize(pipeLogOutput2);
}
pipeLogOutput2.close();
// pipelog3: 11 without pipedata
@@ -99,12 +99,12 @@ public class BufferedPipeDataQueueTest {
pipeDataQueue.clear();
Assert.assertFalse(pipeLogDir.exists());
} catch (Exception e) {
- Assert.fail();
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
}
}
/** Try to take data from a new pipe. Expect to wait indefinitely if no data
offer. */
- // TODO: 抛出NPE
@Test
public void testTake() {
BufferedPipeDataQueue pipeDataQueue = new
BufferedPipeDataQueue(pipeLogDir.getPath());
@@ -142,7 +142,7 @@ public class BufferedPipeDataQueueTest {
Thread.currentThread().interrupt();
}
});
- pipeDataQueue.offer(new TsFilePipeData(null, 0));
+ pipeDataQueue.offer(new TsFilePipeData("", 0));
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
@@ -244,7 +244,7 @@ public class BufferedPipeDataQueueTest {
BufferedPipeDataQueue pipeDataQueue = new
BufferedPipeDataQueue(pipeLogDir.getPath());
Assert.assertEquals(1, pipeDataQueue.getCommitSerialNumber());
Assert.assertEquals(10, pipeDataQueue.getLastMaxSerialNumber());
- PipeData offerPipeData = new TsFilePipeData(null, 11);
+ PipeData offerPipeData = new TsFilePipeData("", 11);
pipeDataList.add(offerPipeData);
pipeDataQueue.offer(offerPipeData);
diff --git
a/server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java
b/server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java
new file mode 100644
index 0000000..8e4d4d0
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/newsync/transport/TransportServiceTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.security.MessageDigest;
+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.setParentDirPath(fileDir.getAbsolutePath());
+ 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 {
+ MessageDigest messageDigest1 = MessageDigest.getInstance("SHA-256");
+ MessageDigest messageDigest2 = MessageDigest.getInstance("SHA-256");
+ BufferedInputStream fir = new BufferedInputStream(new
FileInputStream(firFile));
+ BufferedInputStream sec = new BufferedInputStream(new
FileInputStream(secFile));
+ // To compare the length and hash of the files.
+ Assert.assertEquals(fir.available(), sec.available());
+ byte[] firstBytes = new byte[1024];
+ byte[] secondBytes = new byte[1024];
+ int length = -1;
+ while ((length = fir.read(firstBytes)) != -1) {
+ Assert.assertEquals(length, sec.read(secondBytes));
+ messageDigest1.update(firstBytes, 0, length);
+ messageDigest2.update(secondBytes, 0, length);
+ }
+ fir.close();
+ sec.close();
+ Assert.assertArrayEquals(messageDigest1.digest(),
messageDigest2.digest());
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+}
diff --git a/thrift-sync/src/main/thrift/transport.thrift
b/thrift-sync/src/main/thrift/transport.thrift
index 3eccde3..c84452c 100644
--- a/thrift-sync/src/main/thrift/transport.thrift
+++ b/thrift-sync/src/main/thrift/transport.thrift
@@ -30,10 +30,12 @@ struct IdentityInfo{
1:required string address
// Sender needs to tell receiver its identity.
- 2:required string uuid
+ 2:required string pipeName
+ 3:required i64 createTime
// The version of sender and receiver need to be the same.
- 3:required string version
+ 4:required string version
+
}
enum Type {
@@ -43,6 +45,20 @@ enum Type {
FILE
}
+enum RequestType {
+ CREATE,
+ START,
+ STOP,
+ DROP,
+ HEARTBEAT
+}
+
+enum ResponseType {
+ INFO,
+ WARN,
+ ERROR
+}
+
struct MetaInfo{
// The type of the pipeData in sending.
1:required Type type
@@ -55,12 +71,14 @@ struct MetaInfo{
}
struct SyncRequest{
- 1:required i32 code
- 2:required string msg
+ 1:required RequestType type
+ 2:required string pipeName
+ 3:required string remoteIp
+ 4:required i64 createTime
}
struct SyncResponse{
- 1:required i32 code
+ 1:required ResponseType type
2:required string msg
}
