This is an automated email from the ASF dual-hosted git repository.
lta pushed a commit to branch reimpl_sync
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/reimpl_sync by this push:
new 99e577c complete sync receiver end
99e577c is described below
commit 99e577cbd14dde5837d43f8a0ce212d740fffd0a
Author: lta <[email protected]>
AuthorDate: Thu Aug 22 17:00:11 2019 +0800
complete sync receiver end
---
example/tsfile/pom.xml | 51 ++-
.../iotdb/db/sync/receiver/load/FileLoader.java | 33 ++
.../iotdb/db/sync/receiver/load/IFileLoader.java | 6 +-
.../receiver/recover/ISyncReceiverLogAnalyzer.java | 1 +
.../sync/receiver/recover/ISyncReceiverLogger.java | 30 ++
.../sync/receiver/recover/SyncReceiverLogger.java | 50 +++
.../db/sync/receiver/transfer/SyncServiceImpl.java | 422 ++++++++-------------
.../apache/iotdb/db/sync/sender/conf/Constans.java | 18 +-
.../db/sync/sender/conf/SyncSenderConfig.java | 18 +-
.../sync/sender/transfer/DataTransferManager.java | 114 +++---
.../sync/sender/transfer/IDataTransferManager.java | 2 +-
.../iotdb/db/sync/sender/SingleClientSyncTest.java | 2 +-
service-rpc/src/main/thrift/sync.thrift | 15 +-
13 files changed, 377 insertions(+), 385 deletions(-)
diff --git a/example/tsfile/pom.xml b/example/tsfile/pom.xml
index 1dab0e0..0ae2822 100644
--- a/example/tsfile/pom.xml
+++ b/example/tsfile/pom.xml
@@ -19,32 +19,31 @@
under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
- <modelVersion>4.0.0</modelVersion>
- <parent>
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-examples</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
<groupId>org.apache.iotdb</groupId>
- <artifactId>iotdb-examples</artifactId>
+ <artifactId>tsfile-example</artifactId>
<version>0.9.0-SNAPSHOT</version>
- <relativePath>../pom.xml</relativePath>
- </parent>
- <groupId>org.apache.iotdb</groupId>
- <artifactId>tsfile-example</artifactId>
- <version>0.9.0-SNAPSHOT</version>
- <name>tsfile-example</name>
- <!-- FIXME change it to the project's website -->
- <url>http://www.example.com</url>
- <dependencies>
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <version>4.11</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.iotdb</groupId>
- <artifactId>tsfile</artifactId>
- <version>0.9.0-SNAPSHOT</version>
- </dependency>
- </dependencies>
+ <name>tsfile-example</name>
+ <!-- FIXME change it to the project's website -->
+ <url>http://www.example.com</url>
+ <dependencies>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.11</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>tsfile</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ </dependency>
+ </dependencies>
</project>
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java
new file mode 100644
index 0000000..c09b0ec
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java
@@ -0,0 +1,33 @@
+package org.apache.iotdb.db.sync.receiver.load;
+
+import java.io.File;
+
+public class FileLoader implements IFileLoader {
+
+ private FileLoader(){
+
+ }
+
+ public static FileLoader getInstance(){
+ return FileLoaderHolder.INSTANCE;
+ }
+
+ @Override
+ public void addDeletedFileName(String sgName, File deletedFile) {
+
+ }
+
+ @Override
+ public void addTsfile(String sgName, File tsfile) {
+
+ }
+
+
+ private static class FileLoaderHolder{
+ private static final FileLoader INSTANCE = new FileLoader();
+
+ private static FileLoader getInstance(){
+ return INSTANCE;
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/IFileLoader.java
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/IFileLoader.java
index aaa3ff7..c2d55b0 100644
---
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/IFileLoader.java
+++
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/IFileLoader.java
@@ -18,8 +18,12 @@
*/
package org.apache.iotdb.db.sync.receiver.load;
+import java.io.File;
+
public interface IFileLoader {
- void loadTsFiles();
+ void addDeletedFileName(String sgName, File deletedFile);
+
+ void addTsfile(String sgName, File tsfile);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/ISyncReceiverLogAnalyzer.java
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/ISyncReceiverLogAnalyzer.java
index 5d6351e..8015fb2 100644
---
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/ISyncReceiverLogAnalyzer.java
+++
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/ISyncReceiverLogAnalyzer.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.sync.receiver.recover;
public interface ISyncReceiverLogAnalyzer {
+
void recover();
void scanLogger(String path);
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/ISyncReceiverLogger.java
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/ISyncReceiverLogger.java
index 9fd3109..8bfa200 100644
---
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/ISyncReceiverLogger.java
+++
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/ISyncReceiverLogger.java
@@ -18,6 +18,36 @@
*/
package org.apache.iotdb.db.sync.receiver.recover;
+import java.io.File;
+import java.io.IOException;
+
public interface ISyncReceiverLogger {
+ /**
+ * Start sync deleted files name
+ * @throws IOException
+ */
+ void startSyncDeletedFilesName() throws IOException;
+
+ /**
+ * After a deleted file name is synced to the receiver end, record it in
sync log.
+ * @param file the deleted tsfile
+ * @throws IOException
+ */
+ void finishSyncDeletedFileName(File file) throws IOException;
+
+ /**
+ * Start sync new tsfiles
+ * @throws IOException
+ */
+ void startSyncTsFiles() throws IOException;
+
+ /**
+ *
+ * After a new tsfile is synced to the receiver end, record it in sync log.
+ * @param file new tsfile
+ * @throws IOException
+ */
+ void finishSyncTsfile(File file) throws IOException;
+ void close() throws IOException;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogger.java
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogger.java
new file mode 100644
index 0000000..9a8cc96
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogger.java
@@ -0,0 +1,50 @@
+package org.apache.iotdb.db.sync.receiver.recover;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+
+public class SyncReceiverLogger implements ISyncReceiverLogger {
+
+ public static final String SYNC_DELETED_FILE_NAME_START = "sync deleted file
names start";
+ public static final String SYNC_TSFILE_START = "sync tsfile start";
+ private BufferedWriter bw;
+
+ public SyncReceiverLogger(File logFile) throws IOException {
+ bw = new BufferedWriter(new FileWriter(logFile));
+ }
+
+ @Override
+ public void startSyncDeletedFilesName() throws IOException {
+ bw.write(SYNC_DELETED_FILE_NAME_START);
+ bw.newLine();
+ bw.flush();
+ }
+
+ @Override
+ public void finishSyncDeletedFileName(File file) throws IOException {
+ bw.write(file.getAbsolutePath());
+ bw.newLine();
+ bw.flush();
+ }
+
+ @Override
+ public void startSyncTsFiles() throws IOException {
+ bw.write(SYNC_TSFILE_START);
+ bw.newLine();
+ bw.flush();
+ }
+
+ @Override
+ public void finishSyncTsfile(File file) throws IOException {
+ bw.write(file.getAbsolutePath());
+ bw.newLine();
+ bw.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ bw.close();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java
index 61ff2ec..e2149b6 100644
---
a/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java
+++
b/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java
@@ -20,7 +20,6 @@ package org.apache.iotdb.db.sync.receiver.transfer;
import java.io.BufferedReader;
import java.io.File;
-import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
@@ -28,52 +27,32 @@ import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.security.MessageDigest;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
import org.apache.commons.io.FileUtils;
-import org.apache.commons.lang3.StringUtils;
import org.apache.iotdb.db.concurrent.ThreadName;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
-import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.DiskSpaceInsufficientException;
import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.exception.PathErrorException;
-import org.apache.iotdb.db.exception.ProcessorException;
-import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.metadata.MetadataConstant;
import org.apache.iotdb.db.metadata.MetadataOperationType;
-import org.apache.iotdb.db.qp.executor.QueryProcessExecutor;
-import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.sync.receiver.load.FileLoader;
+import org.apache.iotdb.db.sync.receiver.recover.SyncReceiverLogger;
import org.apache.iotdb.db.sync.sender.conf.Constans;
import org.apache.iotdb.db.utils.FilePathUtils;
import org.apache.iotdb.db.utils.SyncUtils;
import org.apache.iotdb.service.sync.thrift.ResultStatus;
-import org.apache.iotdb.service.sync.thrift.SyncDataStatus;
import org.apache.iotdb.service.sync.thrift.SyncService;
-import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetaData;
-import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
-import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadata;
-import org.apache.iotdb.tsfile.file.metadata.TsDeviceMetadataIndex;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.read.ReadOnlyTsFile;
-import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
-import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.Path;
-import org.apache.iotdb.tsfile.read.common.RowRecord;
-import org.apache.iotdb.tsfile.read.expression.QueryExpression;
-import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -82,67 +61,72 @@ public class SyncServiceImpl implements SyncService.Iface {
private static final Logger logger =
LoggerFactory.getLogger(SyncServiceImpl.class);
- private static final StorageEngine STORAGE_GROUP_MANAGER =
StorageEngine.getInstance();
- /**
- * Metadata manager
- **/
- private static final MManager metadataManger = MManager.getInstance();
+ private static final MManager METADATA_MANGER = MManager.getInstance();
- private static final String SYNC_SERVER = Constans.SYNC_RECEIVER;
+ private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- /**
- * String means storage group,List means the set of new files(path) in local
IoTDB and String
- * means path of new Files
- **/
- private ThreadLocal<Map<String, List<String>>> fileNodeMap = new
ThreadLocal<>();
+ private ThreadLocal<String> syncFolderPath = new ThreadLocal<>();
- /**
- * Total num of files that needs to be loaded
- */
- private ThreadLocal<Integer> fileNum = new ThreadLocal<>();
+ private ThreadLocal<String> syncDataPath = new ThreadLocal<>();
- /**
- * IoTDB config
- **/
- private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private ThreadLocal<String> currentSG = new ThreadLocal<>();
- /**
- * IoTDB data directory
- **/
- private String baseDir = config.getBaseDir();
+ private ThreadLocal<SyncReceiverLogger> syncLog = new ThreadLocal<>();
- /**
- * IoTDB multiple bufferWrite directory
- **/
- private String[] bufferWritePaths = config.getDataDirs();
+ private ThreadLocal<String> senderIp = new ThreadLocal<>();
- /**
- * The path to store metadata file of sender
- */
- private ThreadLocal<String> schemaFromSenderPath = new ThreadLocal<>();
+ private ThreadLocal<File> currentFile = new ThreadLocal<>();
- /**
- * Sync folder path of server
- **/
- private ThreadLocal<String> syncFolderPath = new ThreadLocal<>();
+ private ThreadLocal<FileChannel> currentFileWriter = new ThreadLocal<>();
+
+ private ThreadLocal<MessageDigest> messageDigest = new ThreadLocal<>();
/**
- * Sync data path of server
+ * Verify IP address of sender
*/
- private ThreadLocal<String> syncDataPath = new ThreadLocal<>();
+ @Override
+ public ResultStatus check(String ipAddress) {
+ Thread.currentThread().setName(ThreadName.SYNC_SERVER.getName());
+ if (SyncUtils.verifyIPSegment(config.getIpWhiteList(), ipAddress)) {
+ senderIp.set(ipAddress);
+ if (checkRecovery()) {
+ return getSuccessResult();
+ } else {
+ return getErrorResult("Receiver is processing data from previous sync
tasks");
+ }
+ } else {
+ return getErrorResult(
+ "Sender IP is not in the white list of receiver IP and
synchronization tasks are not allowed.");
+ }
+ }
- private ThreadLocal<String> currentSG = new ThreadLocal<>();
+ private boolean checkRecovery() {
+ return true;
+ }
+ @Override
+ public ResultStatus startSync() {
+ try {
+ initPath();
+ currentSG.remove();
+ syncLog.set(new SyncReceiverLogger(new File(syncDataPath.get(),
Constans.SYNC_LOG_NAME)));
+ return getSuccessResult();
+ } catch (DiskSpaceInsufficientException | IOException e) {
+ logger.error("Can not receiver data from sender", e);
+ return getErrorResult(e.getMessage());
+ }
+ }
/**
- * Verify IP address of sender
+ * Init file path and clear data if last sync process failed.
*/
- @Override
- public ResultStatus checkIdentity(String ipAddress) {
- Thread.currentThread().setName(ThreadName.SYNC_SERVER.getName());
- initPath();
- return SyncUtils.verifyIPSegment(config.getIpWhiteList(), ipAddress) ?
ResultStatus.SUCCESS
- : ResultStatus.FAILURE;
+ private void initPath() throws DiskSpaceInsufficientException {
+ String dataDir =
DirectoryManager.getInstance().getNextFolderForSequenceFile();
+ syncFolderPath
+ .set(FilePathUtils.regularizePath(dataDir) + Constans.SYNC_RECEIVER +
File.separatorChar
+ + senderIp.get());
+ syncDataPath
+ .set(syncFolderPath.get() + File.separatorChar +
Constans.RECEIVER_DATA_FOLDER_NAME);
}
/**
@@ -152,121 +136,106 @@ public class SyncServiceImpl implements
SyncService.Iface {
public ResultStatus init(String storageGroup) {
logger.info("Sync process starts to receive data of storage group {}",
storageGroup);
currentSG.set(storageGroup);
- fileNum.set(0);
- fileNodeMap.set(new HashMap<>());
try {
- FileUtils.deleteDirectory(new File(syncDataPath));
+ syncLog.get().startSyncDeletedFilesName();
} catch (IOException e) {
- logger.error("cannot delete directory {} ", syncFolderPath);
- return ResultStatus.FAILURE;
+ logger.error("Can not init sync process", e);
+ return getErrorResult(e.getMessage());
}
- for (String bufferWritePath : bufferWritePaths) {
- bufferWritePath = FilePathUtils.regularizePath(bufferWritePath);
- String backupPath = bufferWritePath + SYNC_SERVER + File.separator;
- File backupDirectory = new File(backupPath, this.uuid.get());
- if (backupDirectory.exists() && backupDirectory.list().length != 0) {
- try {
- FileUtils.deleteDirectory(backupDirectory);
- } catch (IOException e) {
- logger.error("cannot delete directory {} ", syncFolderPath);
- return ResultStatus.FAILURE;
- }
- }
- }
- return ResultStatus.SUCCESS;
+ return getSuccessResult();
}
- /**
- * Init file path and clear data if last sync process failed.
- */
- private void initPath() {
- baseDir = FilePathUtils.regularizePath(baseDir);
- syncFolderPath = baseDir + SYNC_SERVER + File.separatorChar +
this.uuid.get();
- syncDataPath = syncFolderPath + File.separatorChar +
Constans.DATA_SNAPSHOT_NAME;
- schemaFromSenderPath
- .set(syncFolderPath + File.separator + MetadataConstant.METADATA_LOG);
+ @Override
+ public ResultStatus syncDeletedFileName(String fileName) throws TException {
+ try {
+ syncLog.get().finishSyncDeletedFileName(new File(fileName));
+ FileLoader.getInstance().addDeletedFileName(currentSG.get(), new
File(fileName));
+ } catch (IOException e) {
+ logger.error("Can not sync deleted file", e);
+ return getErrorResult(
+ String.format("Can not sync deleted file %s because %s", fileName,
e.getMessage()));
+ }
+ return getSuccessResult();
}
- /**
- * Acquire schema from sender
- *
- * @param status: FINIFSH_STATUS, SUCCESS_STATUS or PROCESSING_STATUS.
status = FINISH_STATUS :
- * finish receiving schema file, start to sync schema. status =
PROCESSING_STATUS : the schema
- * file has not received completely.SUCCESS_STATUS: load metadata.
- */
@Override
- public String syncSchema(String md5, ByteBuffer schema, SyncDataStatus
status) {
- String md5OfReceiver = Boolean.toString(Boolean.TRUE);
- if (status == SyncDataStatus.SUCCESS_STATUS) {
- /** sync metadata, include storage group and timeseries **/
- return Boolean.toString(loadMetadata());
- } else if (status == SyncDataStatus.PROCESSING_STATUS) {
- File file = new File(schemaFromSenderPath.get());
- if (!file.getParentFile().exists()) {
- try {
- file.getParentFile().mkdirs();
- file.createNewFile();
- } catch (IOException e) {
- logger.error("Cannot make schema file {}.", file.getPath(), e);
- md5OfReceiver = Boolean.toString(Boolean.FALSE);
- }
- }
- try (FileOutputStream fos = new FileOutputStream(file, true);
- FileChannel channel = fos.getChannel()) {
- channel.write(schema);
- } catch (Exception e) {
- logger.error("Cannot insert data to file {}.", file.getPath(), e);
- md5OfReceiver = Boolean.toString(Boolean.FALSE);
+ public ResultStatus initSyncData(String filename) throws TException {
+ try {
+ File file;
+ if (currentSG.get() == null) {
+ file = new File(syncDataPath.get(), filename);
+ } else {
+ file = new File(syncDataPath.get(), currentSG.get() +
File.separatorChar + filename);
}
- } else {
- try (FileInputStream fis = new
FileInputStream(schemaFromSenderPath.get())) {
- MessageDigest md = MessageDigest.getInstance("MD5");
- byte[] buffer = new byte[Constans.DATA_CHUNK_SIZE];
- int n;
- while ((n = fis.read(buffer)) != -1) {
- md.update(buffer, 0, n);
- }
- md5OfReceiver = (new BigInteger(1, md.digest())).toString(16);
- if (!md5.equals(md5OfReceiver)) {
- FileUtils.forceDelete(new File(schemaFromSenderPath.get()));
- }
- } catch (Exception e) {
- logger.error("Receiver cannot generate md5 {}",
schemaFromSenderPath.get(), e);
+ currentFile.set(file);
+ if (!file.getParentFile().exists()) {
+ file.getParentFile().mkdirs();
}
+ currentFileWriter.set(new FileOutputStream(file).getChannel());
+ syncLog.get().startSyncTsFiles();
+ } catch (IOException e) {
+ logger.error("Can not init sync resource for file {}", filename, e);
+ return getErrorResult(
+ String.format("Can not init sync resource for file %s because %s",
filename,
+ e.getMessage()));
}
- return md5OfReceiver;
+ return getSuccessResult();
}
@Override
- public String checkDataMD5(String md5) throws TException {
- return null;
+ public ResultStatus syncData(ByteBuffer buff) {
+ try {
+ currentFileWriter.get().write(buff);
+ messageDigest.get().update(buff);
+ } catch (IOException e) {
+ logger.error("Can not sync data for file {}",
currentFile.get().getAbsoluteFile(), e);
+ return getErrorResult(String
+ .format("Can not sync data for file %s because %s",
currentFile.get().getName(),
+ e.getMessage()));
+ }
+ return getSuccessResult();
}
@Override
- public void endSync() throws TException {
-
+ public ResultStatus checkDataMD5(String md5OfSender) throws TException {
+ String md5OfReceiver = (new BigInteger(1,
messageDigest.get().digest())).toString(16);
+ try {
+ currentFileWriter.get().close();
+ if (!md5OfSender.equals(md5OfReceiver)) {
+ FileUtils.forceDelete(currentFile.get());
+ currentFileWriter.set(new
FileOutputStream(currentFile.get()).getChannel());
+ } else {
+ if
(currentFile.get().getName().endsWith(MetadataConstant.METADATA_LOG)) {
+ loadMetadata();
+ } else {
+ if
(!currentFile.get().getName().endsWith(TsFileResource.RESOURCE_SUFFIX)) {
+ syncLog.get().finishSyncTsfile(currentFile.get());
+ FileLoader.getInstance().addTsfile(currentSG.get(),
currentFile.get());
+ }
+ }
+ }
+ } catch (IOException e) {
+ logger.error("Can not check data MD5 for file {}",
currentFile.get().getAbsoluteFile(), e);
+ return getErrorResult(String
+ .format("Can not check data MD5 for file %s because %s",
currentFile.get().getName(),
+ e.getMessage()));
+ }
+ return new ResultStatus(true, null, md5OfReceiver);
}
- /**
- * Load metadata from sender
- */
private boolean loadMetadata() {
- if (new File(schemaFromSenderPath.get()).exists()) {
+ if (currentFile.get().exists()) {
try (BufferedReader br = new BufferedReader(
- new java.io.FileReader(schemaFromSenderPath.get()))) {
+ new java.io.FileReader(currentFile.get()))) {
String metadataOperation;
while ((metadataOperation = br.readLine()) != null) {
operation(metadataOperation);
}
} catch (FileNotFoundException e) {
- logger.error("Cannot read the file {}.",
- schemaFromSenderPath.get(), e);
- return false;
- } catch (IOException e) {
- /** multiple insert schema, ignore it **/
- } catch (Exception e) {
- logger.error("Parse metadata operation failed.", e);
+ logger.error("Cannot read the file {}.",
currentFile.get().getAbsoluteFile(), e);
return false;
+ } catch (IOException | MetadataErrorException | PathErrorException e) {
+ // multiple insert schema, ignore it.
}
}
return true;
@@ -289,151 +258,56 @@ public class SyncServiceImpl implements
SyncService.Iface {
kv = args[k].split("=");
props.put(kv[0], kv[1]);
}
- metadataManger.addPathToMTree(new Path(args[1]),
TSDataType.deserialize(Short.valueOf(args[2])),
- TSEncoding.deserialize(Short.valueOf(args[3])),
- CompressionType.deserialize(Short.valueOf(args[4])),
- props);
+ METADATA_MANGER
+ .addPathToMTree(new Path(args[1]),
TSDataType.deserialize(Short.valueOf(args[2])),
+ TSEncoding.deserialize(Short.valueOf(args[3])),
+ CompressionType.deserialize(Short.valueOf(args[4])),
+ props);
break;
case MetadataOperationType.DELETE_PATH_FROM_MTREE:
- metadataManger.deletePaths(Collections.singletonList(new
Path(args[1])));
+ METADATA_MANGER.deletePaths(Collections.singletonList(new
Path(args[1])));
break;
case MetadataOperationType.SET_STORAGE_LEVEL_TO_MTREE:
- metadataManger.setStorageLevelToMTree(args[1]);
+ METADATA_MANGER.setStorageLevelToMTree(args[1]);
break;
case MetadataOperationType.ADD_A_PTREE:
- metadataManger.addAPTree(args[1]);
+ METADATA_MANGER.addAPTree(args[1]);
break;
case MetadataOperationType.ADD_A_PATH_TO_PTREE:
- metadataManger.addPathToPTree(args[1]);
+ METADATA_MANGER.addPathToPTree(args[1]);
break;
case MetadataOperationType.DELETE_PATH_FROM_PTREE:
- metadataManger.deletePathFromPTree(args[1]);
+ METADATA_MANGER.deletePathFromPTree(args[1]);
break;
case MetadataOperationType.LINK_MNODE_TO_PTREE:
- metadataManger.linkMNodeToPTree(args[1], args[2]);
+ METADATA_MANGER.linkMNodeToPTree(args[1], args[2]);
break;
case MetadataOperationType.UNLINK_MNODE_FROM_PTREE:
- metadataManger.unlinkMNodeFromPTree(args[1], args[2]);
+ METADATA_MANGER.unlinkMNodeFromPTree(args[1], args[2]);
break;
default:
logger.error("Unrecognizable command {}", cmd);
}
}
- /**
- * Start receiving tsfile from sender
- *
- * @param status status = SUCCESS_STATUS : finish receiving one tsfile
status = PROCESSING_STATUS
- * : tsfile has not received completely.
- */
- @Override
- public String syncData(String md5OfSender, List<String> filePathSplit,
- ByteBuffer dataToReceive, SyncDataStatus status) {
- String md5OfReceiver = Boolean.toString(Boolean.TRUE);
- FileChannel channel;
- /** Recombination File Path **/
- String filePath = StringUtils.join(filePathSplit, File.separatorChar);
- syncDataPath = FilePathUtils.regularizePath(syncDataPath);
- filePath = syncDataPath + filePath;
- if (status == SyncDataStatus.PROCESSING_STATUS) { // there are still data
stream to add
- File file = new File(filePath);
- if (!file.getParentFile().exists()) {
- try {
- file.getParentFile().mkdirs();
- file.createNewFile();
- } catch (IOException e) {
- logger.error("cannot make file {}", file.getPath(), e);
- md5OfReceiver = Boolean.toString(Boolean.FALSE);
- }
- }
- try (FileOutputStream fos = new FileOutputStream(file, true)) {// append
new data
- channel = fos.getChannel();
- channel.write(dataToReceive);
- } catch (IOException e) {
- logger.error("cannot insert data to file {}", file.getPath(), e);
- md5OfReceiver = Boolean.toString(Boolean.FALSE);
-
- }
- } else { // all data in the same file has received successfully
- try (FileInputStream fis = new FileInputStream(filePath)) {
- MessageDigest md = MessageDigest.getInstance("MD5");
- byte[] buffer = new byte[Constans.DATA_CHUNK_SIZE];
- int n;
- while ((n = fis.read(buffer)) != -1) {
- md.update(buffer, 0, n);
- }
- md5OfReceiver = (new BigInteger(1, md.digest())).toString(16);
- if (md5OfSender.equals(md5OfReceiver)) {
- fileNum.set(fileNum.get() + 1);
-
- logger.info(String.format("Receiver has received %d files from
sender", fileNum.get()));
- } else {
- FileUtils.forceDelete(new File(filePath));
- }
- } catch (Exception e) {
- logger.error("Receiver cannot generate md5 {}", filePath, e);
- }
- }
- return md5OfReceiver;
- }
-
-
- @Override
- public boolean load() {
- try {
- getFileNodeInfo();
- loadData();
- } catch (Exception e) {
- logger.error("fail to load data", e);
- return false;
- }
- return true;
- }
-
- /**
- * Open all tsfile reader and cache
- */
- private Map<String, ReadOnlyTsFile> openReaders(String filePath,
List<String> overlapFiles)
- throws IOException {
- Map<String, ReadOnlyTsFile> tsfileReaders = new HashMap<>();
- tsfileReaders.put(filePath, new ReadOnlyTsFile(new
TsFileSequenceReader(filePath)));
- for (String overlapFile : overlapFiles) {
- tsfileReaders.put(overlapFile, new ReadOnlyTsFile(new
TsFileSequenceReader(overlapFile)));
- }
- return tsfileReaders;
- }
-
- /**
- * Close all tsfile reader
- */
- private void closeReaders(Map<String, ReadOnlyTsFile> readers) throws
IOException {
- for (ReadOnlyTsFile tsfileReader : readers.values()) {
- tsfileReader.close();
- }
- }
-
- /**
- * Release threadLocal variable resources
- */
@Override
- public void cleanUp() {
- fileNum.remove();
- fileNodeMap.remove();
- schemaFromSenderPath.remove();
+ public ResultStatus endSync() throws TException {
try {
- FileUtils.deleteDirectory(new File(syncFolderPath));
+ syncLog.get().close();
+ new File(syncDataPath.get(), Constans.SYNC_END).createNewFile();
} catch (IOException e) {
- logger.error("can not delete directory {}", syncFolderPath, e);
+ logger.error("Can not end sync", e);
+ return getErrorResult(String.format("Can not end sync because {}",
e.getMessage()));
}
- logger.info("Synchronization has finished!");
+ return getSuccessResult();
}
- public Map<String, List<String>> getFileNodeMap() {
- return fileNodeMap.get();
+ private ResultStatus getSuccessResult() {
+ return new ResultStatus(true, null, null);
}
- public void setFileNodeMap(Map<String, List<String>> fileNodeMap) {
- this.fileNodeMap.set(fileNodeMap);
+ private ResultStatus getErrorResult(String errorMsg) {
+ return new ResultStatus(false, errorMsg, null);
}
}
\ No newline at end of file
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/sender/conf/Constans.java
b/server/src/main/java/org/apache/iotdb/db/sync/sender/conf/Constans.java
index dffa1e0..576b416 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/sender/conf/Constans.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/conf/Constans.java
@@ -28,8 +28,14 @@ public class Constans {
public static final String SYNC_RECEIVER = "sync-receiver";
public static final String MESSAGE_DIGIT_NAME = "MD5";
+
public static final String SYNC_DIR_NAME_SEPARATOR = "_";
+ /**
+ * Split data file , block size at each transmission
+ **/
+ public static final int DATA_CHUNK_SIZE = 64 * 1024 * 1024;
+
// sender section
public static final String LOCK_FILE_NAME = "sync_lock";
@@ -45,11 +51,6 @@ public class Constans {
public static final String SYNC_LOG_NAME = "sync.log";
/**
- * Split data file , block size at each transmission
- **/
- public static final int DATA_CHUNK_SIZE = 64 * 1024 * 1024;
-
- /**
* Maximum try when syncing the same file to receiver fails.
*/
public static final int MAX_SYNC_FILE_TRY = 5;
@@ -64,4 +65,11 @@ public class Constans {
public static final long SYNC_MONITOR_PERIOD =
CONFIG.getSyncPeriodInSecond();
+ // receiver section
+
+ public static final String RECEIVER_DATA_FOLDER_NAME = "data";
+
+ public static final String SYNC_END = "sync.end";
+
+
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/sender/conf/SyncSenderConfig.java
b/server/src/main/java/org/apache/iotdb/db/sync/sender/conf/SyncSenderConfig.java
index a076257..d6028ca 100644
---
a/server/src/main/java/org/apache/iotdb/db/sync/sender/conf/SyncSenderConfig.java
+++
b/server/src/main/java/org/apache/iotdb/db/sync/sender/conf/SyncSenderConfig.java
@@ -28,7 +28,7 @@ public class SyncSenderConfig {
private int syncPeriodInSecond = 10;
- private String senderPath;
+ private String senderFolderPath;
private String lockFilePath;
@@ -40,11 +40,11 @@ public class SyncSenderConfig {
* Update paths based on data directory
*/
public void update(String dataDirectory) {
- senderPath = dataDirectory + File.separatorChar + Constans.SYNC_SENDER +
File.separatorChar +
+ senderFolderPath = dataDirectory + File.separatorChar +
Constans.SYNC_SENDER + File.separatorChar +
getSyncReceiverName();
- lockFilePath = senderPath + File.separatorChar + Constans.LOCK_FILE_NAME;
- lastFileInfo = senderPath + File.separatorChar +
Constans.LAST_LOCAL_FILE_NAME;
- snapshotPath = senderPath + File.separatorChar +
Constans.DATA_SNAPSHOT_NAME;
+ lockFilePath = senderFolderPath + File.separatorChar +
Constans.LOCK_FILE_NAME;
+ lastFileInfo = senderFolderPath + File.separatorChar +
Constans.LAST_LOCAL_FILE_NAME;
+ snapshotPath = senderFolderPath + File.separatorChar +
Constans.DATA_SNAPSHOT_NAME;
if(!new File(snapshotPath).exists()){
new File(snapshotPath).mkdirs();
}
@@ -74,12 +74,12 @@ public class SyncSenderConfig {
this.syncPeriodInSecond = syncPeriodInSecond;
}
- public String getSenderPath() {
- return senderPath;
+ public String getSenderFolderPath() {
+ return senderFolderPath;
}
- public void setSenderPath(String senderPath) {
- this.senderPath = senderPath;
+ public void setSenderFolderPath(String senderFolderPath) {
+ this.senderFolderPath = senderFolderPath;
}
public String getLockFilePath() {
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/DataTransferManager.java
b/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/DataTransferManager.java
index 12803f4..ce67c27 100644
---
a/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/DataTransferManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/DataTransferManager.java
@@ -1,19 +1,15 @@
/**
- * 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
+ * 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
+ * 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
+ * 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.sync.sender.transfer;
@@ -36,10 +32,8 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -100,10 +94,6 @@ public class DataTransferManager implements
IDataTransferManager {
private Map<String, Set<File>> lastLocalFilesMap;
- private Map<String, Set<File>> successSyncedFilesMap = new HashMap<>();
-
- private Map<String, Set<File>> successDeletedFilesMap = new HashMap<>();
-
/**
* If true, sync is in execution.
**/
@@ -227,10 +217,8 @@ public class DataTransferManager implements
IDataTransferManager {
// 1. Connect to sync receiver and confirm identity
establishConnection(config.getServerIp(), config.getServerPort());
- if (!confirmIdentity()) {
- logger.error("Sorry, you do not have the permission to connect to sync
receiver.");
- System.exit(1);
- }
+ confirmIdentity();
+ serviceClient.startSync();
// 2. Sync Schema
syncSchema();
@@ -239,6 +227,7 @@ public class DataTransferManager implements
IDataTransferManager {
String[] dataDirs =
IoTDBDescriptor.getInstance().getConfig().getDataDirs();
for (String dataDir : dataDirs) {
logger.info("Start to sync data in data dir {}", dataDir);
+
config.update(dataDir);
syncFileManager.getValidFiles(dataDir);
allSG = syncFileManager.getAllSG();
@@ -267,7 +256,7 @@ public class DataTransferManager implements
IDataTransferManager {
}
private void checkRecovery() {
- new SyncSenderLogAnalyzer(config.getSenderPath()).recover();
+ new SyncSenderLogAnalyzer(config.getSenderFolderPath()).recover();
}
@Override
@@ -284,10 +273,13 @@ public class DataTransferManager implements
IDataTransferManager {
}
@Override
- public boolean confirmIdentity() throws SyncConnectionException {
+ public void confirmIdentity() throws SyncConnectionException {
try {
- return
serviceClient.checkIdentity(InetAddress.getLocalHost().getHostAddress())
- == ResultStatus.SUCCESS;
+ ResultStatus status =
serviceClient.check(InetAddress.getLocalHost().getHostAddress());
+ if (!status.success) {
+ throw new SyncConnectionException(
+ "The receiver rejected the synchronization task because " +
status.errorMsg);
+ }
} catch (Exception e) {
logger.error("Cannot confirm identity with the receiver.");
throw new SyncConnectionException(e);
@@ -335,8 +327,9 @@ public class DataTransferManager implements
IDataTransferManager {
if (cntLine++ == BATCH_LINE) {
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
- if (serviceClient.syncData(buffToSend) == ResultStatus.FAILURE) {
- logger.error("Receiver failed to receive metadata, retry.");
+ ResultStatus status = serviceClient.syncData(buffToSend);
+ if (!status.success) {
+ logger.error("Receiver failed to receive metadata because {},
retry.", status.errorMsg);
return false;
}
cntLine = 0;
@@ -345,8 +338,9 @@ public class DataTransferManager implements
IDataTransferManager {
if (bos.size() != 0) {
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
- if (serviceClient.syncData(buffToSend) == ResultStatus.FAILURE) {
- logger.error("Receiver failed to receive metadata, retry.");
+ ResultStatus status = serviceClient.syncData(buffToSend);
+ if (!status.success) {
+ logger.error("Receiver failed to receive metadata because {},
retry.", status.errorMsg);
return false;
}
}
@@ -363,8 +357,8 @@ public class DataTransferManager implements
IDataTransferManager {
* Check MD5 of schema to make sure that the receiver receives the schema
correctly
*/
private boolean checkMD5ForSchema(String md5OfSender) throws TException {
- String md5OfReceiver = serviceClient.checkDataMD5(md5OfSender);
- if (md5OfSender.equals(md5OfReceiver)) {
+ ResultStatus status = serviceClient.checkDataMD5(md5OfSender);
+ if (status.success && md5OfSender.equals(status.msg)) {
logger.info("Receiver has received schema successfully, retry.");
return true;
} else {
@@ -405,14 +399,14 @@ public class DataTransferManager implements
IDataTransferManager {
try {
syncStatus = true;
syncLog = new SyncSenderLogger(getSchemaLogFile());
- successSyncedFilesMap = new HashMap<>();
- successDeletedFilesMap = new HashMap<>();
for (String sgName : allSG) {
+ lastLocalFilesMap.putIfAbsent(sgName, new HashSet<>());
syncLog = new SyncSenderLogger(getSyncLogFile());
try {
- if (serviceClient.init(sgName) == ResultStatus.FAILURE) {
- throw new SyncConnectionException("unable init receiver");
+ ResultStatus status = serviceClient.init(sgName);
+ if (!status.success) {
+ throw new SyncConnectionException("Unable init receiver because "
+ status.errorMsg);
}
} catch (TException | SyncConnectionException e) {
throw new SyncConnectionException("Unable to connect to receiver",
e);
@@ -444,9 +438,10 @@ public class DataTransferManager implements
IDataTransferManager {
logger.info("Start to sync names of deleted files in storage group {}",
sgName);
for (File file : deletedFilesName) {
try {
- serviceClient.syncDeletedFileName(file.getName());
- successDeletedFilesMap.get(sgName).add(file);
- syncLog.finishSyncDeletedFileName(file);
+ if (serviceClient.syncDeletedFileName(file.getName()).success) {
+ lastLocalFilesMap.get(sgName).add(file);
+ syncLog.finishSyncDeletedFileName(file);
+ }
} catch (TException e) {
logger.error("Can not sync deleted file name {}, skip it.", file);
}
@@ -469,9 +464,10 @@ public class DataTransferManager implements
IDataTransferManager {
File snapshotFile = null;
try {
snapshotFile = makeFileSnapshot(tsfile);
- syncSingleFile(snapshotFile);
+ // firstly sync .restore file, then sync tsfile
syncSingleFile(new File(snapshotFile, TsFileResource.RESOURCE_SUFFIX));
- successSyncedFilesMap.get(sgName).add(tsfile);
+ syncSingleFile(snapshotFile);
+ lastLocalFilesMap.get(sgName).add(tsfile);
syncLog.finishSyncTsfile(tsfile);
logger.info("Task of synchronization has completed {}/{}.", cnt,
toBeSyncFiles.size());
} catch (IOException e) {
@@ -534,9 +530,10 @@ public class DataTransferManager implements
IDataTransferManager {
md.update(buffer, 0, dataLength);
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
- if (serviceClient.syncData(buffToSend) == ResultStatus.FAILURE) {
- logger.info("Receiver failed to receive data from {}, retry.",
- snapshotFile.getAbsoluteFile());
+ ResultStatus status = serviceClient.syncData(buffToSend);
+ if (!status.success) {
+ logger.info("Receiver failed to receive data from {} because {},
retry.",
+ status.errorMsg, snapshotFile.getAbsoluteFile());
continue outer;
}
}
@@ -544,10 +541,12 @@ public class DataTransferManager implements
IDataTransferManager {
// the file is sent successfully
String md5OfSender = (new BigInteger(1, md.digest())).toString(16);
- String md5OfReceiver = serviceClient.checkDataMD5(md5OfSender);
- if (md5OfSender.equals(md5OfReceiver)) {
+ ResultStatus status = serviceClient.checkDataMD5(md5OfSender);
+ if (status.success && md5OfSender.equals(status.msg)) {
logger.info("Receiver has received {} successfully.",
snapshotFile.getAbsoluteFile());
break;
+ } else {
+ logger.error("MD5 check of tsfile {} failed, retry",
snapshotFile.getAbsoluteFile());
}
}
} catch (IOException | TException | NoSuchAlgorithmException e) {
@@ -556,17 +555,10 @@ public class DataTransferManager implements
IDataTransferManager {
}
private void endSync() {
- // 1. Organize current local files based on sync result
- for (Entry<String, Set<File>> entry : lastLocalFilesMap.entrySet()) {
- entry.getValue()
- .removeAll(successDeletedFilesMap.getOrDefault(entry.getKey(), new
HashSet<>()));
- entry.getValue()
- .removeAll(successSyncedFilesMap.getOrDefault(entry.getKey(), new
HashSet<>()));
- }
File currentLocalFile = getCurrentLogFile();
File lastLocalFile = new File(config.getLastFileInfo());
- // 2. Write file list to currentLocalFile
+ // 1. Write file list to currentLocalFile
try (BufferedWriter bw = new BufferedWriter(new
FileWriter(currentLocalFile))) {
for (Set<File> currentLocalFiles : lastLocalFilesMap.values()) {
for (File file : currentLocalFiles) {
@@ -579,24 +571,24 @@ public class DataTransferManager implements
IDataTransferManager {
logger.error("Can not clear sync log {}",
lastLocalFile.getAbsoluteFile(), e);
}
- // 3. Rename currentLocalFile to lastLocalFile
+ // 2. Rename currentLocalFile to lastLocalFile
lastLocalFile.deleteOnExit();
currentLocalFile.renameTo(lastLocalFile);
- // 4. delete snapshot directory
+ // 3. delete snapshot directory
try {
FileUtils.deleteDirectory(new File(config.getSnapshotPath()));
} catch (IOException e) {
logger.error("Can not clear snapshot directory {}",
config.getSnapshotPath(), e);
}
- // 5. delete sync log file
+ // 4. delete sync log file
getSyncLogFile().deleteOnExit();
}
private File getSchemaPosFile() {
- return new File(config.getSenderPath(), Constans.SCHEMA_POS_FILE_NAME);
+ return new File(config.getSenderFolderPath(),
Constans.SCHEMA_POS_FILE_NAME);
}
private File getSchemaLogFile() {
@@ -610,11 +602,11 @@ public class DataTransferManager implements
IDataTransferManager {
}
private File getSyncLogFile() {
- return new File(config.getSenderPath(), Constans.SYNC_LOG_NAME);
+ return new File(config.getSenderFolderPath(), Constans.SYNC_LOG_NAME);
}
private File getCurrentLogFile() {
- return new File(config.getSenderPath(), Constans.CURRENT_LOCAL_FILE_NAME);
+ return new File(config.getSenderFolderPath(),
Constans.CURRENT_LOCAL_FILE_NAME);
}
public void setConfig(SyncSenderConfig config) {
diff --git
a/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/IDataTransferManager.java
b/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/IDataTransferManager.java
index bff3f2f..98d6d14 100644
---
a/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/IDataTransferManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/IDataTransferManager.java
@@ -42,7 +42,7 @@ public interface IDataTransferManager {
/**
* Confirm identity, the receiver will check whether the sender has
synchronization privileges.
*/
- boolean confirmIdentity() throws SyncConnectionException, IOException;
+ void confirmIdentity() throws SyncConnectionException, IOException;
/**
* Sync schema file to receiver before all data to be synced.
diff --git
a/server/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
b/server/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
index 7a06281..4a7ac7e 100644
---
a/server/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
@@ -159,7 +159,7 @@ public class SingleClientSyncTest {
}
public void setConfig() {
- config.setSenderPath(
+ config.setSenderFolderPath(
config.getDataDirectory() + SYNC_CLIENT + File.separator +
Constans.UUID_FILE_NAME);
config.setLastFileInfo(
config.getDataDirectory() + SYNC_CLIENT + File.separator +
Constans.LAST_LOCAL_FILE_NAME);
diff --git a/service-rpc/src/main/thrift/sync.thrift
b/service-rpc/src/main/thrift/sync.thrift
index 6fc394a..2064c4e 100755
--- a/service-rpc/src/main/thrift/sync.thrift
+++ b/service-rpc/src/main/thrift/sync.thrift
@@ -22,18 +22,19 @@ typedef i32 int
typedef i16 short
typedef i64 long
-enum ResultStatus {
- SUCCESS,
- FAILURE,
- BUSY
+struct ResultStatus{
+ bool success,
+ string errorMsg,
+ string msg
}
service SyncService{
- ResultStatus checkIdentity(1:string address)
+ ResultStatus check(1:string address)
+ ResultStatus startSync();
ResultStatus init(1:string storageGroupName)
ResultStatus syncDeletedFileName(1:string fileName)
ResultStatus initSyncData(1:string filename)
ResultStatus syncData(1:binary buff)
- string checkDataMD5(1:string md5)
- void endSync()
+ ResultStatus checkDataMD5(1:string md5)
+ ResultStatus endSync()
}
\ No newline at end of file
