You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by lt...@apache.org on 2019/03/18 02:55:55 UTC
[incubator-iotdb] branch Sync-Reconstruct created (now 7471c48)
This is an automated email from the ASF dual-hosted git repository.
lta pushed a change to branch Sync-Reconstruct
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at 7471c48 test sync function and modify the implementation of singleton
This branch includes the following new commits:
new 96f3710 rename postback to sync and reconstruct sync client
new 2f2e1d8 reconstruct thrift sync service
new ab830f9 reconstruct sync server
new 239174c test sync features
new 7471c48 test sync function and modify the implementation of singleton
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 05/05: test sync function and modify the
implementation of singleton
Posted by lt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lta pushed a commit to branch Sync-Reconstruct
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 7471c48f426f89c241be3504c068ec92d1501e7d
Author: lta <li...@163.com>
AuthorDate: Mon Mar 18 10:35:57 2019 +0800
test sync function and modify the implementation of singleton
---
iotdb/iotdb/conf/iotdb-sync-client.properties | 11 ++-
.../org/apache/iotdb/db/sync/conf/Constans.java | 1 +
.../iotdb/db/sync/conf/SyncSenderConfig.java | 21 ++--
.../iotdb/db/sync/conf/SyncSenderDescriptor.java | 6 +-
.../iotdb/db/sync/sender/FileSenderImpl.java | 110 ++++++++++++++-------
5 files changed, 90 insertions(+), 59 deletions(-)
diff --git a/iotdb/iotdb/conf/iotdb-sync-client.properties b/iotdb/iotdb/conf/iotdb-sync-client.properties
index bb70b2f..479d1c0 100644
--- a/iotdb/iotdb/conf/iotdb-sync-client.properties
+++ b/iotdb/iotdb/conf/iotdb-sync-client.properties
@@ -17,16 +17,19 @@
# under the License.
#
+# Sync server port address
server_ip=127.0.0.1
-# PostBack server port address
+
+# Sync client port
server_port=5555
-# PostBack client port
-client_port=6666
+
# The cycle time of post data back to receiver, the unit of time is second
upload_cycle_in_seconds=600
-# Set bufferWrite data absolute path of IoTDB
+
+# Set bufferWrite data absolute path of IoTDB
# It needs to be set with iotdb_schema_directory, they have to belong to the same IoTDB
# iotdb_bufferWrite_directory = D:\\iotdb\\data\\data\\settled
+
# Set schema file absolute path of IoTDB
# It needs to be set with iotdb_bufferWrite_directory, they have to belong to the same IoTDB
# iotdb_schema_directory = D:\\iotdb\\data\\system\\schema\\mlog.txt
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
index 739bbbf..8e0556c 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
@@ -30,6 +30,7 @@ public class Constans {
public static final String SYNC_CLIENT = "sync-client";
public static final String SYNC_SERVER = "sync-server";
+ public static final String LOCK_FILE_NAME = "sync-lock";
public static final String UUID_FILE_NAME = "uuid.txt";
public static final String LAST_LOCAL_FILE_NAME = "last_local_files.txt";
public static final String DATA_SNAPSHOT_NAME = "data-snapshot";
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
index 05b4659..3fcdb15 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
@@ -30,15 +30,14 @@ public class SyncSenderConfig {
private String[] bufferwriteDirectory = IoTDBDescriptor.getInstance().getConfig()
.getBufferWriteDirs();
private String dataDirectory = IoTDBDescriptor.getInstance().getConfig().getDataDir();
+ private String lockFilePath;
private String uuidPath;
private String lastFileInfo;
private String[] snapshotPaths;
private String schemaPath;
private String serverIp = "127.0.0.1";
private int serverPort = 5555;
- private int clientPort = 6666;
private int uploadCycleInSeconds = 10;
- private boolean clearEnable = false;
public void init() {
String metadataDirPath = IoTDBDescriptor.getInstance().getConfig().getMetadataDir();
@@ -51,6 +50,8 @@ public class SyncSenderConfig {
&& dataDirectory.charAt(dataDirectory.length() - 1) != File.separatorChar) {
dataDirectory += File.separatorChar;
}
+ lockFilePath =
+ dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.LOCK_FILE_NAME;
uuidPath = dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.UUID_FILE_NAME;
lastFileInfo =
dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME;
@@ -133,14 +134,6 @@ public class SyncSenderConfig {
this.serverPort = serverPort;
}
- public int getClientPort() {
- return clientPort;
- }
-
- public void setClientPort(int clientPort) {
- this.clientPort = clientPort;
- }
-
public int getUploadCycleInSeconds() {
return uploadCycleInSeconds;
}
@@ -149,11 +142,11 @@ public class SyncSenderConfig {
this.uploadCycleInSeconds = uploadCycleInSeconds;
}
- public boolean getClearEnable() {
- return clearEnable;
+ public String getLockFilePath() {
+ return lockFilePath;
}
- public void setClearEnable(boolean clearEnable) {
- this.clearEnable = clearEnable;
+ public void setLockFilePath(String lockFilePath) {
+ this.lockFilePath = lockFilePath;
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
index 9ca52c1..301e13f 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
@@ -89,9 +89,6 @@ public class SyncSenderDescriptor {
conf.setServerIp(properties.getProperty("server_ip", conf.getServerIp()));
conf.setServerPort(Integer
.parseInt(properties.getProperty("server_port", Integer.toString(conf.getServerPort()))));
-
- conf.setClientPort(Integer
- .parseInt(properties.getProperty("client_port", Integer.toString(conf.getClientPort()))));
conf.setUploadCycleInSeconds(Integer.parseInt(properties
.getProperty("upload_cycle_in_seconds",
Integer.toString(conf.getUploadCycleInSeconds()))));
@@ -106,7 +103,8 @@ public class SyncSenderDescriptor {
conf.setUuidPath(
dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.UUID_FILE_NAME);
conf.setLastFileInfo(
- dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME);
+ dataDirectory + Constans.SYNC_CLIENT + File.separatorChar
+ + Constans.LAST_LOCAL_FILE_NAME);
String[] iotdbBufferwriteDirectory = conf.getBufferwriteDirectory();
String[] snapshots = new String[conf.getBufferwriteDirectory().length];
for (int i = 0; i < conf.getBufferwriteDirectory().length; i++) {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
index 0279033..8c95331 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
@@ -10,7 +10,6 @@
* 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
@@ -25,11 +24,11 @@ import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
+import java.io.RandomAccessFile;
import java.math.BigInteger;
import java.net.InetAddress;
-import java.net.ServerSocket;
-import java.net.Socket;
import java.nio.ByteBuffer;
+import java.nio.channels.FileLock;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -149,7 +148,7 @@ public class FileSenderImpl implements FileSender {
throws InterruptedException, IOException, SyncConnectionException {
Thread.currentThread().setName(ThreadName.SYNC_CLIENT.getName());
FileSenderImpl fileSenderImpl = new FileSenderImpl();
- fileSenderImpl.verifyPort();
+ fileSenderImpl.verifySingleton();
fileSenderImpl.startMonitor();
fileSenderImpl.timedTask();
}
@@ -198,23 +197,19 @@ public class FileSenderImpl implements FileSender {
}
}
- syncStatus = true;
-
- // 2. Connect to sync server and Confirm Identity
- establishConnection(config.getServerIp(), config.getServerPort());
- if (!confirmIdentity(config.getUuidPath())) {
- LOGGER.error("Sorry, you do not have the permission to connect to sync receiver.");
- syncStatus = false;
- return;
- }
-
- // 3. Acquire valid files and check
+ // 2. Acquire valid files and check
fileManager.init();
validAllFiles = fileManager.getValidAllFiles();
currentLocalFiles = fileManager.getCurrentLocalFiles();
if (SyncUtils.isEmpty(validAllFiles)) {
LOGGER.info("There has no file to sync !");
- syncStatus = false;
+ return;
+ }
+
+ // 3. Connect to sync server and Confirm Identity
+ establishConnection(config.getServerIp(), config.getServerPort());
+ if (!confirmIdentity(config.getUuidPath())) {
+ LOGGER.error("Sorry, you do not have the permission to connect to sync receiver.");
return;
}
@@ -223,6 +218,8 @@ public class FileSenderImpl implements FileSender {
validFileSnapshot.put(entry.getKey(), makeFileSnapshot(entry.getValue()));
}
+ syncStatus = true;
+
// 5. Sync schema
syncSchema();
@@ -463,32 +460,71 @@ public class FileSenderImpl implements FileSender {
}
/**
- * The method is to verify whether the client port is bind or not, ensuring that only one client
- * is running.
+ * The method is to verify whether the client lock file is locked or not, ensuring that only one
+ * client is running.
+ */
+ private void verifySingleton() throws IOException {
+ File lockFile = new File(config.getLockFilePath());
+ if (!lockFile.getParentFile().exists()) {
+ lockFile.getParentFile().mkdirs();
+ }
+ if (!lockFile.exists()) {
+ lockFile.createNewFile();
+ }
+ if (!lockInstance(config.getLockFilePath())) {
+ LOGGER.error("Sync client is running.");
+ System.exit(1);
+ }
+// try {
+// Socket socket = new Socket("localhost", config.getClientPort());
+// socket.close();
+// LOGGER.error("Sync client has been started!");
+// System.exit(1);
+// } catch (IOException e) {
+// try (ServerSocket listenerSocket = new ServerSocket(config.getClientPort())) {
+// Thread listener = new Thread(() -> {
+// try {
+// while (true) {
+// listenerSocket.accept();
+// }
+// } catch (IOException e2) {
+// LOGGER.error("Unable to listen to port{}", config.getClientPort(), e2);
+// }
+// });
+// listener.start();
+// } catch (IOException e1) {
+// LOGGER.error("Unable to listen to port{}", config.getClientPort());
+// throw new IOException();
+// }
+// }
+ }
+
+ /**
+ * Try to lock lockfile. if failed, it means that sync client has benn started.
+ *
+ * @param lockFile path of lockfile
*/
- private void verifyPort() throws IOException {
+ private static boolean lockInstance(final String lockFile) {
try {
- Socket socket = new Socket("localhost", config.getClientPort());
- socket.close();
- LOGGER.error("Sync client has been started!");
- System.exit(0);
- } catch (IOException e) {
- try (ServerSocket listenerSocket = new ServerSocket(config.getClientPort())) {
- Thread listener = new Thread(() -> {
- while (true) {
- try {
- listenerSocket.accept();
- } catch (IOException e2) {
- LOGGER.error("Unable to listen to port{}", config.getClientPort(), e2);
- }
+ final File file = new File(lockFile);
+ final RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw");
+ final FileLock fileLock = randomAccessFile.getChannel().tryLock();
+ if (fileLock != null) {
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ try {
+ fileLock.release();
+ randomAccessFile.close();
+ file.delete();
+ } catch (Exception e) {
+ LOGGER.error("Unable to remove lock file: {}", lockFile, e);
}
- });
- listener.start();
- } catch (IOException e1) {
- LOGGER.error("Unable to listen to port{}", config.getClientPort());
- throw new IOException();
+ }));
+ return true;
}
+ } catch (Exception e) {
+ LOGGER.error("Unable to create and/or lock file: {}", lockFile, e);
}
+ return false;
}
private static class InstanceHolder {
[incubator-iotdb] 03/05: reconstruct sync server
Posted by lt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lta pushed a commit to branch Sync-Reconstruct
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit ab830f9b15060898ea040f473353e36c818c5a51
Author: lta <li...@163.com>
AuthorDate: Sun Mar 17 16:12:02 2019 +0800
reconstruct sync server
---
.../db/engine/filenode/FileNodeProcessor.java | 10 +-
.../org/apache/iotdb/db/sync/conf/Constans.java | 5 +-
.../iotdb/db/sync/conf/SyncSenderConfig.java | 6 +-
.../iotdb/db/sync/conf/SyncSenderDescriptor.java | 6 +-
.../iotdb/db/sync/receiver/ServerServiceImpl.java | 232 +++++++++++++--------
.../apache/iotdb/db/sync/sender/FileManager.java | 2 +-
.../iotdb/db/sync/sender/FileSenderImpl.java | 12 +-
.../iotdb/db/sync/sender/FileManagerTest.java | 2 +-
service-rpc/src/main/thrift/sync.thrift | 6 +-
9 files changed, 164 insertions(+), 117 deletions(-)
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
index 3fddd8d..be775a5 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.engine.filenode;
+import static java.time.ZonedDateTime.ofInstant;
+
import java.io.File;
import java.io.IOException;
import java.nio.file.FileSystems;
@@ -41,7 +43,6 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -102,8 +103,6 @@ import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static java.time.ZonedDateTime.ofInstant;
-
public class FileNodeProcessor extends Processor implements IStatistic {
private static final String WARN_NO_SUCH_OVERFLOWED_FILE = "Can not find any tsfile which"
@@ -838,9 +837,8 @@ public class FileNodeProcessor extends Processor implements IStatistic {
appendFile.getFilePath()));
}
if (!originFile.renameTo(targetFile)) {
- LOGGER.warn("File renaming failed when appending new file. Origin: {}, target: {}",
- originFile.getPath(),
- targetFile.getPath());
+ LOGGER.warn("File renaming failed when appending new file. Origin: {}, Target: {}",
+ originFile.getPath(), targetFile.getPath());
}
// append the new tsfile
this.newFileNodes.add(appendFile);
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
index 31e0ad4..cc2581a 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
@@ -27,11 +27,12 @@ public class Constans {
}
public static final String CONFIG_NAME = "iotdb-sync-client.properties";
- public static final String SYNC = "sync";
+ public static final String SYNC_CLIENT = "sync-client";
+ public static final String SYNC_SERVER = "sync-server";
public static final String UUID_FILE_NAME = "uuid.txt";
public static final String LAST_LOCAL_FILE_NAME = "last_local_files.txt";
- public static final String DATA_SNAPSHOT_NAME = "data_snapshot";
+ public static final String DATA_SNAPSHOT_NAME = "data-snapshot";
/**
* Split data file , block size at each transmission
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
index 0def77c..fe33f2e 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
@@ -51,9 +51,9 @@ public class SyncSenderConfig {
&& dataDirectory.charAt(dataDirectory.length() - 1) != File.separatorChar) {
dataDirectory += File.separatorChar;
}
- uuidPath = dataDirectory + Constans.SYNC + File.separatorChar + Constans.UUID_FILE_NAME;
+ uuidPath = dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.UUID_FILE_NAME;
lastFileInfo =
- dataDirectory + Constans.SYNC + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME;
+ dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME;
snapshotPaths = new String[iotdbBufferwriteDirectory.length];
for (int i = 0; i < iotdbBufferwriteDirectory.length; i++) {
iotdbBufferwriteDirectory[i] = new File(iotdbBufferwriteDirectory[i]).getAbsolutePath();
@@ -62,7 +62,7 @@ public class SyncSenderConfig {
!= File.separatorChar) {
iotdbBufferwriteDirectory[i] = iotdbBufferwriteDirectory[i] + File.separatorChar;
}
- snapshotPaths[i] = iotdbBufferwriteDirectory[i] + Constans.SYNC + File.separatorChar
+ snapshotPaths[i] = iotdbBufferwriteDirectory[i] + Constans.SYNC_CLIENT + File.separatorChar
+ Constans.DATA_SNAPSHOT_NAME
+ File.separatorChar;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
index 04a5e7d..c04336f 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
@@ -104,9 +104,9 @@ public class SyncSenderDescriptor {
dataDirectory += File.separatorChar;
}
conf.setUuidPath(
- dataDirectory + Constans.SYNC + File.separatorChar + Constans.UUID_FILE_NAME);
+ dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.UUID_FILE_NAME);
conf.setLastFileInfo(
- dataDirectory + Constans.SYNC + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME);
+ dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME);
String[] iotdbBufferwriteDirectory = conf.getIotdbBufferwriteDirectory();
String[] snapshots = new String[conf.getIotdbBufferwriteDirectory().length];
for (int i = 0; i < conf.getIotdbBufferwriteDirectory().length; i++) {
@@ -115,7 +115,7 @@ public class SyncSenderDescriptor {
!= File.separatorChar) {
iotdbBufferwriteDirectory[i] = iotdbBufferwriteDirectory[i] + File.separatorChar;
}
- snapshots[i] = iotdbBufferwriteDirectory[i] + Constans.SYNC + File.separatorChar
+ snapshots[i] = iotdbBufferwriteDirectory[i] + Constans.SYNC_CLIENT + File.separatorChar
+ Constans.DATA_SNAPSHOT_NAME
+ File.separatorChar;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
index 435c20b..9674767 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
@@ -35,6 +35,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Objects;
import java.util.Set;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -47,9 +48,11 @@ import org.apache.iotdb.db.exception.MetadataArgsErrorException;
import org.apache.iotdb.db.exception.PathErrorException;
import org.apache.iotdb.db.exception.ProcessorException;
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.OverflowQPExecutor;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.sync.conf.Constans;
import org.apache.iotdb.db.utils.SyncUtils;
import org.apache.iotdb.service.sync.thrift.SyncDataStatus;
import org.apache.iotdb.service.sync.thrift.SyncService;
@@ -77,27 +80,57 @@ import org.slf4j.LoggerFactory;
public class ServerServiceImpl implements SyncService.Iface {
private static final Logger logger = LoggerFactory.getLogger(ServerServiceImpl.class);
+
private static final FileNodeManager fileNodeManager = FileNodeManager.getInstance();
+ /**
+ * Metadata manager
+ **/
private static final MManager metadataManger = MManager.getInstance();
- private static final String POSTBACK = "sync";
+
+ private static final String SYNC_SERVER = Constans.SYNC_SERVER;
+
private ThreadLocal<String> uuid = new ThreadLocal<>();
- // String means Storage Group,List means the set of new Files(AbsulutePath) in local IoTDB
- // String means AbsulutePath of new Files
+ /**
+ * 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<>();
- // Map String1 means timeseries String2 means AbsulutePath of new Files, long means startTime
+ /**
+ * Map String1 means timeseries String2 means path of new Files, long means startTime
+ **/
private ThreadLocal<Map<String, Map<String, Long>>> fileNodeStartTime = new ThreadLocal<>();
- // Map String1 means timeseries String2 means AbsulutePath of new Files, long means endTime
+ /**
+ * Map String1 means timeseries String2 means path of new Files, long means endTime
+ **/
private ThreadLocal<Map<String, Map<String, Long>>> fileNodeEndTime = new ThreadLocal<>();
+
+ /**
+ * Total num of files that needs to be loaded
+ */
private ThreadLocal<Integer> fileNum = new ThreadLocal<>();
+
+ /**
+ * IoTDB ioTDBConfig
+ **/
+ private IoTDBConfig ioTDBConfig = IoTDBDescriptor.getInstance().getConfig();
+
+ /**
+ * IoTDB data directory
+ **/
+ private String dataPath = ioTDBConfig.getDataDir();
+
+ /**
+ * IoTDB multiple bufferWrite directory
+ **/
+ private String[] bufferWritePaths = ioTDBConfig.getBufferWriteDirs();
+
+ /**
+ * The path to store metadata file of sender
+ */
private ThreadLocal<String> schemaFromSenderPath = new ThreadLocal<>();
- private IoTDBConfig tsfileDBconfig = IoTDBDescriptor.getInstance().getConfig();
- private String postbackPath;
- // Absolute seriesPath of IoTDB data directory
- private String dataPath =
- new File(tsfileDBconfig.getDataDir()).getAbsolutePath() + File.separator;
- // Absolute paths of IoTDB bufferWrite directory
- private String[] bufferWritePaths = tsfileDBconfig.getBufferWriteDirs();
- private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+
+ /** Sync path of server **/
+ private String syncPath;
/**
* Init threadLocal variable
@@ -105,9 +138,7 @@ public class ServerServiceImpl implements SyncService.Iface {
@Override
public void init(String storageGroup) {
if (logger.isInfoEnabled()) {
- logger.info(
- "IoTDB post back receiver: sync process starts to receive data of storage group {}",
- storageGroup);
+ logger.info("Sync process starts to receive data of storage group {}", storageGroup);
}
fileNum.set(0);
fileNodeMap.set(new HashMap<>());
@@ -119,73 +150,88 @@ public class ServerServiceImpl implements SyncService.Iface {
* Verify IP address of sender
*/
@Override
- public boolean getUUID(String uuid, String ipAddress) throws TException {
+ public boolean checkIdentity(String uuid, String ipAddress) throws TException {
this.uuid.set(uuid);
- postbackPath = dataPath + POSTBACK + File.separator;
- schemaFromSenderPath.set(postbackPath + this.uuid.get() + File.separator + "mlog.txt");
- if (new File(postbackPath + this.uuid.get()).exists()
- && new File(postbackPath + this.uuid.get()).list().length != 0) {
+ initPath();
+ return SyncUtils.verifyIPSegment(ioTDBConfig.getIpWhiteList(), ipAddress);
+ }
+
+ /**
+ * Init file path and clear data if last sync process failed.
+ */
+ private void initPath() throws TException {
+ if (dataPath.length() > 0 && dataPath.charAt(dataPath.length() - 1) != File.separatorChar) {
+ dataPath = dataPath + File.separatorChar;
+ }
+ syncPath = dataPath + SYNC_SERVER + File.separator;
+ schemaFromSenderPath
+ .set(syncPath + this.uuid.get() + File.separator + MetadataConstant.METADATA_LOG);
+ File syncFileDirectory = new File(syncPath,this.uuid.get());
+ if (syncFileDirectory.exists()
+ && Objects.requireNonNull(syncFileDirectory.list()).length != 0) {
try {
- SyncUtils.deleteFile(new File(postbackPath + this.uuid.get()));
+ SyncUtils.deleteFile(syncFileDirectory);
} catch (IOException e) {
+ logger.error("Cannot clear useless metadata file.");
throw new TException(e);
}
}
for (String bufferWritePath : bufferWritePaths) {
- String backupPath = bufferWritePath + POSTBACK + File.separator;
- if (new File(backupPath + this.uuid.get()).exists()
- && new File(backupPath + this.uuid.get()).list().length != 0) {
- // if does not exist, it means that the last time sync failed, clear uuid
- // data and receive the data again
+ if (bufferWritePath.length() > 0
+ && bufferWritePath.charAt(bufferWritePath.length() - 1) != File.separatorChar) {
+ bufferWritePath = bufferWritePath + File.separatorChar;
+ }
+ String backupPath = bufferWritePath + SYNC_SERVER + File.separator;
+ File backupDirectory = new File(backupPath, this.uuid.get());
+ if (backupDirectory.exists() && Objects.requireNonNull(backupDirectory.list()).length != 0) {
+ /** if does not exist, it means that the last time sync failed, clear data in the uuid directory and receive the data again **/
try {
- SyncUtils.deleteFile(new File(backupPath + this.uuid.get()));
+ SyncUtils.deleteFile(backupDirectory);
} catch (IOException e) {
+ logger.error("Cannot clear useless backup data file");
throw new TException(e);
}
}
}
- return SyncUtils.verifyIPSegment(config.getIpWhiteList(), ipAddress);
}
-
/**
- * Get schema from sender
+ * Acquire schema from sender
*
- * @param status: 0 or 1. status = 0 : finish receiving schema file, start to insert schema to
- * IoTDB through jdbc status = 1 : the schema file has not received completely.
+ * @param status: SUCCESS_STATUS or PROCESSING_STATUS. status = SUCCESS_STATUS : finish receiving schema file, start to sync schema.
+ * status = SUCCESS_STATUS : the schema file has not received completely.
*/
@Override
- public void getSchema(ByteBuffer schema, SyncDataStatus status) {
+ public void syncSchema(ByteBuffer schema, SyncDataStatus status) {
if (status == SyncDataStatus.SUCCESS_STATUS) {
/** sync metadata, include storage group and timeseries **/
- syncMetadata();
+ loadMetadata();
} else {
File file = new File(schemaFromSenderPath.get());
if (!file.getParentFile().exists()) {
try {
file.getParentFile().mkdirs();
if (!file.createNewFile()) {
- logger.error("IoTDB post back receiver: cannot create file {}",
- file.getAbsoluteFile());
+ logger.error("Cannot create file {}", file.getPath());
}
} catch (IOException e) {
- logger.error("IoTDB post back receiver: cannot make schema file.", e);
+ logger.error("Cannot make schema file {}.", file.getPath(),e);
}
}
try (FileOutputStream fos = new FileOutputStream(file, true);
FileChannel channel = fos.getChannel()) {
channel.write(schema);
} catch (Exception e) {
- logger.error("IoTDB post back receiver: cannot write data to file.", e);
+ logger.error("Cannot write data to file {}.", file.getPath(),e);
}
}
}
/**
- * Sync metadata with sender
+ * Load metadata from sender
*/
- private void syncMetadata() {
+ private void loadMetadata() {
if (new File(schemaFromSenderPath.get()).exists()) {
try (BufferedReader br = new BufferedReader(
new java.io.FileReader(schemaFromSenderPath.get()))) {
@@ -194,12 +240,12 @@ public class ServerServiceImpl implements SyncService.Iface {
operation(metadataOperation);
}
} catch (FileNotFoundException e) {
- logger.error("IoTDB post back receiver: cannot read the file {}.",
+ logger.error("Cannot read the file {}.",
schemaFromSenderPath.get(), e);
} catch (IOException e) {
- logger.error("IoTDB post back receiver: cannot insert schema to IoTDB.", e);
+ logger.error("Cannot insert schema to IoTDB.", e);
} catch (Exception e) {
- logger.error("IoTDB post back receiver: parse metadata operation failed.", e);
+ logger.error("Parse metadata operation failed.", e);
}
}
}
@@ -255,15 +301,16 @@ public class ServerServiceImpl implements SyncService.Iface {
/**
* Start receiving tsfile from sender
*
- * @param status status = 0 : finish receiving one tsfile status = 1 : a tsfile has not received
+ * @param status status = SUCCESS_STATUS : finish receiving one tsfile status = PROCESSING_STATUS : tsfile has not received
* completely.
*/
@Override
- public String receiveData(String md5OfSender, List<String> filePathSplit,
+ public String syncData(String md5OfSender, List<String> filePathSplit,
ByteBuffer dataToReceive, SyncDataStatus status) {
String md5OfReceiver = "";
StringBuilder filePathBuilder = new StringBuilder();
FileChannel channel;
+ /**Recombination File Path**/
for (int i = 0; i < filePathSplit.size(); i++) {
if (i == filePathSplit.size() - 1) {
filePathBuilder.append(filePathSplit.get(i));
@@ -272,31 +319,33 @@ public class ServerServiceImpl implements SyncService.Iface {
}
}
String filePath = filePathBuilder.toString();
- filePath = postbackPath + uuid.get() + File.separator + filePath;
+ if(syncPath.length() > 0 && syncPath.charAt(syncPath.length()-1)!=File.separatorChar){
+ syncPath = syncPath+File.separatorChar;
+ }
+ filePath = syncPath + uuid.get() + File.separator + 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();
if (!file.createNewFile()) {
- logger.error("IoTDB post back receiver: cannot create file {}", file.getAbsoluteFile());
+ logger.error("cannot create file {}", file.getPath());
}
} catch (IOException e) {
- logger.error("IoTDB post back receiver: cannot make file", e);
+ logger.error("cannot make file {}", file.getPath(), e);
}
}
try (FileOutputStream fos = new FileOutputStream(file, true)) {// append new data
channel = fos.getChannel();
channel.write(dataToReceive);
} catch (IOException e) {
- logger.error("IoTDB post back receiver: cannot write data to file", e);
+ logger.error("cannot write data to file {}",file.getPath(), e);
}
} else { // all data in the same file has received successfully
try (FileInputStream fis = new FileInputStream(filePath)) {
MessageDigest md = MessageDigest.getInstance("MD5");
- int mBufferSize = 8 * 1024 * 1024;
- byte[] buffer = new byte[mBufferSize];
+ byte[] buffer = new byte[Constans.DATA_CHUNK_SIZE];
int n;
while ((n = fis.read(buffer)) != -1) {
md.update(buffer, 0, n);
@@ -305,18 +354,15 @@ public class ServerServiceImpl implements SyncService.Iface {
if (md5OfSender.equals(md5OfReceiver)) {
fileNum.set(fileNum.get() + 1);
if (logger.isInfoEnabled()) {
- logger.info(String
- .format("IoTDB post back receiver : Receiver has received %d files from sender",
- fileNum.get()));
+ logger.info(String.format("Receiver has received %d files from sender", fileNum.get()));
}
} else {
if (!new File(filePath).delete()) {
- logger.error("IoTDB post back receiver : Receiver can not delete file {}",
- new File(filePath).getAbsolutePath());
+ logger.error("Receiver can not delete file {}", new File(filePath).getPath());
}
}
} catch (Exception e) {
- logger.error("IoTDB post back receiver: cannot generate md5", e);
+ logger.error("Receiver cannot generate md5", e);
}
}
return md5OfReceiver;
@@ -328,19 +374,22 @@ public class ServerServiceImpl implements SyncService.Iface {
getFileNodeInfo();
loadData();
try {
- SyncUtils.deleteFile(new File(postbackPath + this.uuid.get()));
+ SyncUtils.deleteFile(new File(syncPath + this.uuid.get()));
} catch (IOException e) {
throw new TException(e);
}
for (String bufferWritePath : bufferWritePaths) {
- String backupPath = bufferWritePath + POSTBACK + File.separator;
- if (new File(backupPath + this.uuid.get()).exists()
- && new File(backupPath + this.uuid.get()).list().length != 0) {
- // if does not exist, it means that the last time sync process failed, clear
- // uuid data and receive the data again
+ if (bufferWritePath.length() > 0
+ && bufferWritePath.charAt(bufferWritePath.length() - 1) != File.separatorChar) {
+ bufferWritePath = bufferWritePath + File.separatorChar;
+ }
+ String backupPath = bufferWritePath + SYNC_SERVER + File.separator;
+ File backupDirectory = new File(backupPath, this.uuid.get());
+ if (backupDirectory.exists() && Objects.requireNonNull(backupDirectory.list()).length != 0) {
try {
- SyncUtils.deleteFile(new File(backupPath + this.uuid.get()));
+ SyncUtils.deleteFile(backupDirectory);
} catch (IOException e) {
+ logger.error("Cannot clear useless backup data file");
throw new TException(e);
}
}
@@ -352,10 +401,9 @@ public class ServerServiceImpl implements SyncService.Iface {
* Get all tsfiles' info which are sent from sender, it is prepare for merging these data
*/
public void getFileNodeInfo() {
- String filePath = postbackPath + uuid.get() + File.separator + "data";
- File root = new File(filePath);
- File[] files = root.listFiles();
- int num = 0;
+ File dataFileRoot = new File(syncPath,this.uuid.get());
+ File[] files = dataFileRoot.listFiles();
+ int processedNum = 0;
for (File storageGroupPB : files) {
List<String> filesPath = new ArrayList<>();
File[] filesSG = storageGroupPB.listFiles();
@@ -364,36 +412,33 @@ public class ServerServiceImpl implements SyncService.Iface {
Map<String, Long> endTimeMap = new HashMap<>();
TsFileSequenceReader reader = null;
try {
- reader = new TsFileSequenceReader(fileTF.getAbsolutePath());
+ reader = new TsFileSequenceReader(fileTF.getPath());
Map<String, TsDeviceMetadataIndex> deviceIdMap = reader.readFileMetadata().getDeviceMap();
Iterator<String> it = deviceIdMap.keySet().iterator();
while (it.hasNext()) {
- String key = it.next(); // key represent device
+ String key = it.next();
TsDeviceMetadataIndex device = deviceIdMap.get(key);
startTimeMap.put(key, device.getStartTime());
endTimeMap.put(key, device.getEndTime());
}
} catch (Exception e) {
- logger.error("IoTDB post back receiver: unable to read tsfile {}",
- fileTF.getAbsolutePath(), e);
+ logger.error("Unable to read tsfile {}", fileTF.getPath(), e);
} finally {
try {
if (reader != null) {
reader.close();
}
} catch (IOException e) {
- logger.error("IoTDB receiver : Cannot close file stream {}",
- fileTF.getAbsolutePath(), e);
+ logger.error("Cannot close file stream {}", fileTF.getPath(), e);
}
}
fileNodeStartTime.get().put(fileTF.getAbsolutePath(), startTimeMap);
fileNodeEndTime.get().put(fileTF.getAbsolutePath(), endTimeMap);
filesPath.add(fileTF.getAbsolutePath());
- num++;
+ processedNum++;
if (logger.isInfoEnabled()) {
logger.info(String
- .format("IoTDB receiver : Getting FileNode Info has complete : %d/%d", num,
- fileNum.get()));
+ .format("Get tsfile info has complete : %d/%d", processedNum, fileNum.get()));
}
fileNodeMap.get().put(storageGroupPB.getName(), filesPath);
}
@@ -406,11 +451,14 @@ public class ServerServiceImpl implements SyncService.Iface {
* directly. If data in the tsfile is old, it has two strategy to merge.It depends on the
* possibility of updating historical data.
*/
- public void loadData() throws TException {
- int num = 0;
+ public void loadData() {
+ if(syncPath.length() > 0 && syncPath.charAt(syncPath.length()-1) != File.separatorChar){
+ syncPath =syncPath + File.separatorChar;
+ }
+ int processedNum = 0;
for (String storageGroup : fileNodeMap.get().keySet()) {
List<String> filesPath = fileNodeMap.get().get(storageGroup);
- // before load extern tsFile, it is necessary to order files in the same SG
+ /** before load external tsFile, it is necessary to order files in the same storage group **/
for (int i = 0; i < filesPath.size(); i++) {
for (int j = i + 1; j < filesPath.size(); j++) {
boolean swapOrNot = false;
@@ -439,7 +487,7 @@ public class ServerServiceImpl implements SyncService.Iface {
Map<String, Long> endTimeMap = fileNodeEndTime.get().get(path);
// create a new fileNode
- String header = postbackPath + uuid.get() + File.separator + "data" + File.separator;
+ String header = syncPath + uuid.get() + File.separator;
String relativePath = path.substring(header.length());
TsFileResource fileNode = new TsFileResource(startTimeMap, endTimeMap,
OverflowChangeType.NO_CHANGE,
@@ -448,7 +496,7 @@ public class ServerServiceImpl implements SyncService.Iface {
try {
if (!fileNodeManager.appendFileToFileNode(storageGroup, fileNode, path)) {
// it is a file with overflow data
- if (config.isUpdate_historical_data_possibility()) {
+ if (ioTDBConfig.isUpdate_historical_data_possibility()) {
loadOldData(path);
} else {
List<String> overlapFiles = fileNodeManager.getOverlapFilesFromFileNode(
@@ -462,13 +510,13 @@ public class ServerServiceImpl implements SyncService.Iface {
}
}
} catch (FileNodeManagerException e) {
- logger.error("IoTDB receiver : Can not load external file ", e);
+ logger.error("Can not load external file ", e);
}
- num++;
+ processedNum++;
if (logger.isInfoEnabled()) {
logger.info(String
- .format("IoTDB receiver : Merging files has completed : %d/%d", num, fileNum.get()));
+ .format("Merging files has completed : %d/%d", processedNum, fileNum.get()));
}
}
}
@@ -477,7 +525,7 @@ public class ServerServiceImpl implements SyncService.Iface {
/**
* Insert all data in the tsfile into IoTDB.
*/
- public void loadOldData(String filePath) throws TException {
+ public void loadOldData(String filePath) {
Set<String> timeseriesSet = new HashSet<>();
TsFileSequenceReader reader = null;
OverflowQPExecutor insertExecutor = new OverflowQPExecutor();
@@ -536,7 +584,7 @@ public class ServerServiceImpl implements SyncService.Iface {
}
}
} catch (IOException e) {
- logger.error("IoTDB receiver can not parse tsfile into SQL", e);
+ logger.error("Receiver can not parse tsfile into SQL", e);
} catch (ProcessorException e) {
logger.error("Meet error while processing non-query.", e);
} finally {
@@ -545,7 +593,7 @@ public class ServerServiceImpl implements SyncService.Iface {
reader.close();
}
} catch (IOException e) {
- logger.error("IoTDB receiver : Cannot close file stream {}", filePath, e);
+ logger.error("Cannot close file stream {}", filePath, e);
}
}
}
@@ -664,14 +712,14 @@ public class ServerServiceImpl implements SyncService.Iface {
}
}
} catch (IOException e) {
- logger.error("IoTDB receiver can not parse tsfile into SQL", e);
+ logger.error("Can not parse tsfile into SQL", e);
} catch (ProcessorException e) {
logger.error("Meet error while processing non-query.", e);
} finally {
try {
reader.close();
} catch (IOException e) {
- logger.error("IoTDB receiver : Cannot close file stream {}", filePath, e);
+ logger.error("Cannot close file stream {}", filePath, e);
}
}
}
@@ -687,7 +735,7 @@ public class ServerServiceImpl implements SyncService.Iface {
fileNodeStartTime.remove();
fileNodeEndTime.remove();
schemaFromSenderPath.remove();
- logger.info("IoTDB post back receiver: the postBack has finished!");
+ logger.info("Synchronization has finished!");
}
public Map<String, List<String>> getFileNodeMap() {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
index f0a3b66..d2d4258 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
@@ -149,7 +149,7 @@ public class FileManager {
}
File[] listFiles = new File(path).listFiles();
for (File storageGroup : listFiles) {
- if (storageGroup.isDirectory() && !storageGroup.getName().equals(Constans.SYNC)) {
+ if (storageGroup.isDirectory() && !storageGroup.getName().equals(Constans.SYNC_CLIENT)) {
if (!currentLocalFiles.containsKey(storageGroup.getName())) {
currentLocalFiles.put(storageGroup.getName(), new HashSet<>());
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
index e8939de..c099c56 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
@@ -332,7 +332,7 @@ public class FileSenderImpl implements FileSender {
}
boolean legalConnection;
try {
- legalConnection = serviceClient.getUUID(uuid,
+ legalConnection = serviceClient.checkIdentity(uuid,
InetAddress.getLocalHost().getHostAddress());
} catch (Exception e) {
LOGGER.error("cannot confirm identity with receiver");
@@ -342,7 +342,7 @@ public class FileSenderImpl implements FileSender {
}
private String generateUUID() {
- return Constans.SYNC + UUID.randomUUID().toString().replaceAll("-", "");
+ return Constans.SYNC_CLIENT + UUID.randomUUID().toString().replaceAll("-", "");
}
/**
@@ -403,7 +403,7 @@ public class FileSenderImpl implements FileSender {
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
serviceClient
- .receiveData(null, filePathSplit, buffToSend, SyncDataStatus.PROCESSING_STATUS);
+ .syncData(null, filePathSplit, buffToSend, SyncDataStatus.PROCESSING_STATUS);
}
bos.close();
}
@@ -418,7 +418,7 @@ public class FileSenderImpl implements FileSender {
// the file is sent successfully
String md5OfSender = (new BigInteger(1, md.digest())).toString(16);
- String md5OfReceiver = serviceClient.receiveData(md5OfSender, filePathSplit,
+ String md5OfReceiver = serviceClient.syncData(md5OfSender, filePathSplit,
null, SyncDataStatus.SUCCESS_STATUS);
if (md5OfSender.equals(md5OfReceiver)) {
LOGGER.info("receiver has received {} successfully.", snapshotFilePath);
@@ -450,11 +450,11 @@ public class FileSenderImpl implements FileSender {
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
// 1 represents there is still schema buffer to send.
- serviceClient.getSchema(buffToSend, SyncDataStatus.PROCESSING_STATUS);
+ serviceClient.syncSchema(buffToSend, SyncDataStatus.PROCESSING_STATUS);
}
bos.close();
// 0 represents the schema file has been transferred completely.
- serviceClient.getSchema(null, SyncDataStatus.SUCCESS_STATUS);
+ serviceClient.syncSchema(null, SyncDataStatus.SUCCESS_STATUS);
} catch (Exception e) {
LOGGER.error("cannot sync schema ", e);
throw new SyncConnectionException(e);
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
index a2dd6c6..a23bdbc 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
@@ -35,7 +35,7 @@ import org.slf4j.LoggerFactory;
public class FileManagerTest {
- public static final String POST_BACK_DIRECTORY_TEST = Constans.SYNC + File.separator;
+ public static final String POST_BACK_DIRECTORY_TEST = Constans.SYNC_CLIENT + File.separator;
public static final String LAST_FILE_INFO_TEST =
POST_BACK_DIRECTORY_TEST + Constans.LAST_LOCAL_FILE_NAME;
public static final String SENDER_FILE_PATH_TEST = POST_BACK_DIRECTORY_TEST + "data";
diff --git a/service-rpc/src/main/thrift/sync.thrift b/service-rpc/src/main/thrift/sync.thrift
index a947507..e139783 100755
--- a/service-rpc/src/main/thrift/sync.thrift
+++ b/service-rpc/src/main/thrift/sync.thrift
@@ -28,9 +28,9 @@ enum SyncDataStatus {
}
service SyncService{
- bool getUUID(1:string uuid, 2:string address)
- void getSchema(1:binary buff, 2:SyncDataStatus status)
- string receiveData(1:string md5, 2:list<string> filename, 3:binary buff, 4:SyncDataStatus status)
+ bool checkIdentity(1:string uuid, 2:string address)
+ void syncSchema(1:binary buff, 2:SyncDataStatus status)
+ string syncData(1:string md5, 2:list<string> filename, 3:binary buff, 4:SyncDataStatus status)
bool load()
void cleanUp()
void init(1:string storageGroup)
[incubator-iotdb] 04/05: test sync features
Posted by lt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lta pushed a commit to branch Sync-Reconstruct
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 239174c6fb68614c6908413eaae245fbff4ade4a
Author: lta <li...@163.com>
AuthorDate: Sun Mar 17 23:42:35 2019 +0800
test sync features
---
iotdb/iotdb/bin/start-sync-client.bat | 4 +-
iotdb/iotdb/bin/start-sync-client.sh | 4 +-
iotdb/iotdb/bin/stop-sync-client.bat | 2 +-
iotdb/iotdb/bin/stop-sync-client.sh | 2 +-
iotdb/iotdb/conf/iotdb-engine.properties | 12 +--
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 20 ++--
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 14 +--
.../db/engine/filenode/FileNodeProcessor.java | 11 ++-
.../org/apache/iotdb/db/sync/conf/Constans.java | 2 +
.../iotdb/db/sync/conf/SyncSenderConfig.java | 24 ++---
.../iotdb/db/sync/conf/SyncSenderDescriptor.java | 11 +--
.../iotdb/db/sync/receiver/ServerManager.java | 13 +--
.../iotdb/db/sync/receiver/ServerServiceImpl.java | 109 ++++++++++-----------
.../apache/iotdb/db/sync/sender/FileManager.java | 12 +--
.../iotdb/db/sync/sender/FileSenderImpl.java | 54 ++++------
.../java/org/apache/iotdb/db/utils/SyncUtils.java | 39 ++++----
.../iotdb/db/sync/sender/FileManagerTest.java | 37 +++----
...stBackTest.java => MultipleClientSyncTest.java} | 8 +-
...PostBackTest.java => SingleClientSyncTest.java} | 53 +++++-----
...CreateDataSender1.java => SyncTestClient1.java} | 6 +-
...CreateDataSender2.java => SyncTestClient2.java} | 6 +-
...CreateDataSender3.java => SyncTestClient3.java} | 8 +-
.../java/org/apache/iotdb/db/sync/test/Utils.java | 2 +-
.../{start-postBackTest.sh => start-sync-test.sh} | 0
.../{stop-postBackTest.sh => stop-sync-test.sh} | 0
25 files changed, 224 insertions(+), 229 deletions(-)
diff --git a/iotdb/iotdb/bin/start-sync-client.bat b/iotdb/iotdb/bin/start-sync-client.bat
index e8b0e85..7d7fae6 100755
--- a/iotdb/iotdb/bin/start-sync-client.bat
+++ b/iotdb/iotdb/bin/start-sync-client.bat
@@ -29,7 +29,7 @@ set IOTDB_CONF=%IOTDB_HOME%\conf
set IOTDB_LOGS=%IOTDB_HOME%\logs
-if NOT DEFINED MAIN_CLASS set MAIN_CLASS=org.apache.iotdb.db.postback.sender.FileSenderImpl
+if NOT DEFINED MAIN_CLASS set MAIN_CLASS=org.apache.iotdb.db.sync.sender.FileSenderImpl
if NOT DEFINED JAVA_HOME goto :err
@REM -----------------------------------------------------------------------------
@@ -46,7 +46,7 @@ set CLASSPATH="%IOTDB_HOME%\lib"
REM For each jar in the IOTDB_HOME lib directory call append to build the CLASSPATH variable.
for %%i in ("%IOTDB_HOME%\lib\*.jar") do call :append "%%i"
-set CLASSPATH=%CLASSPATH%;postBackClient
+set CLASSPATH=%CLASSPATH%;SyncClient
goto okClasspath
:append
diff --git a/iotdb/iotdb/bin/start-sync-client.sh b/iotdb/iotdb/bin/start-sync-client.sh
index eb3e5a4..4d1cfda 100755
--- a/iotdb/iotdb/bin/start-sync-client.sh
+++ b/iotdb/iotdb/bin/start-sync-client.sh
@@ -47,8 +47,8 @@ for f in ${IOTDB_HOME}/lib/*.jar; do
CLASSPATH=${CLASSPATH}":"$f
done
-MAIN_CLASS=org.apache.iotdb.db.postback.sender.FileSenderImpl
+MAIN_CLASS=org.apache.iotdb.db.sync.sender.FileSenderImpl
-"$JAVA" -DIOTDB_HOME=${IOTDB_HOME} -DTSFILE_HOME=${IOTDB_HOME} -DIOTDB_CONF=${IOTDB_CONF} -Dlogback.configurationFile=${IOTDB_CONF}/logback.xml $IOTDB_DERBY_OPTS $IOTDB_JMX_OPTS -Dname=postBackClient -cp "$CLASSPATH" "$MAIN_CLASS"
+"$JAVA" -DIOTDB_HOME=${IOTDB_HOME} -DTSFILE_HOME=${IOTDB_HOME} -DIOTDB_CONF=${IOTDB_CONF} -Dlogback.configurationFile=${IOTDB_CONF}/logback.xml $IOTDB_DERBY_OPTS $IOTDB_JMX_OPTS -Dname=SyncClient -cp "$CLASSPATH" "$MAIN_CLASS"
exit $?
diff --git a/iotdb/iotdb/bin/stop-sync-client.bat b/iotdb/iotdb/bin/stop-sync-client.bat
index d7908ec..698bbbc 100755
--- a/iotdb/iotdb/bin/stop-sync-client.bat
+++ b/iotdb/iotdb/bin/stop-sync-client.bat
@@ -19,5 +19,5 @@
@echo off
-wmic process where (commandline like "%%postBackClient%%" and not name="wmic.exe") delete
+wmic process where (commandline like "%%SyncClient%%" and not name="wmic.exe") delete
rem ps ax | grep -i 'postBackClient' | grep -v grep | awk '{print $1}' | xargs kill -SIGTERM
\ No newline at end of file
diff --git a/iotdb/iotdb/bin/stop-sync-client.sh b/iotdb/iotdb/bin/stop-sync-client.sh
index 9ff0ab7..6e95455 100755
--- a/iotdb/iotdb/bin/stop-sync-client.sh
+++ b/iotdb/iotdb/bin/stop-sync-client.sh
@@ -26,5 +26,5 @@ if [ -z "$PIDS" ]; then
exit 1
else
kill -s TERM $PIDS
- echo "close PostBackClient"
+ echo "close SyncClient"
fi
diff --git a/iotdb/iotdb/conf/iotdb-engine.properties b/iotdb/iotdb/conf/iotdb-engine.properties
index cdca52c..5c7046f 100644
--- a/iotdb/iotdb/conf/iotdb-engine.properties
+++ b/iotdb/iotdb/conf/iotdb-engine.properties
@@ -201,20 +201,20 @@ schema_manager_cache_size=300000
# Generally the default value 4MB is enough.
max_log_entry_size=4194304
-# IoTDB postBack server properties
+# IoTDB sync server properties
# Whether to allow to post back, the default allowed
-is_postback_enable=true
+is_sync_enable=true
-# PostBack server port address
-postback_server_port=5555
+# Sync server port address
+sync_server_port=5555
-# White IP list of Postback client.
+# White IP list of Sync client.
# Please use the form of network segment to present the range of IP, for example: 192.168.0.0/16
# If there are more than one IP segment, please separate them by commas
# The default is to allow all IP to sync
IP_white_list=0.0.0.0/0
-# Choose a postBack strategy of merging historical data:
+# Choose a sync strategy of loading historical data:
#1. It's more likely to update historical data, please choose "true".
#2. It's more likely not to update historical data or you don't know exactly, please choose "false".
update_historical_data_possibility=false
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 815e36b..18d68cd 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -255,11 +255,11 @@ public class IoTDBConfig {
/**
* Is this IoTDB instance a receiver of sync or not.
*/
- private boolean isPostbackEnable = true;
+ private boolean isSyncEnable = true;
/**
* If this IoTDB instance is a receiver of sync, set the server port.
*/
- private int postbackServerPort = 5555;
+ private int syncServerPort = 5555;
/*
* Set the language version when loading file including error information, default value is "EN"
*/
@@ -768,20 +768,20 @@ public class IoTDBConfig {
this.maxLogEntrySize = maxLogEntrySize;
}
- public boolean isPostbackEnable() {
- return isPostbackEnable;
+ public boolean isSyncEnable() {
+ return isSyncEnable;
}
- public void setPostbackEnable(boolean postbackEnable) {
- isPostbackEnable = postbackEnable;
+ public void setSyncEnable(boolean syncEnable) {
+ isSyncEnable = syncEnable;
}
- public int getPostbackServerPort() {
- return postbackServerPort;
+ public int getSyncServerPort() {
+ return syncServerPort;
}
- public void setPostbackServerPort(int postbackServerPort) {
- this.postbackServerPort = postbackServerPort;
+ public void setSyncServerPort(int syncServerPort) {
+ this.syncServerPort = syncServerPort;
}
public String getLanguageVersion() {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 4108a2f..8da75f2 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -190,15 +190,15 @@ public class IoTDBDescriptor {
properties.getProperty("overflow_file_size_threshold",
Long.toString(conf.getOverflowFileSizeThreshold())).trim()));
- conf.setPostbackEnable(Boolean
- .parseBoolean(properties.getProperty("is_postback_enable",
- Boolean.toString(conf.isPostbackEnable()))));
- conf.setPostbackServerPort(Integer
- .parseInt(properties.getProperty("postback_server_port",
- Integer.toString(conf.getPostbackServerPort())).trim()));
+ conf.setSyncEnable(Boolean
+ .parseBoolean(properties.getProperty("is_sync_enable",
+ Boolean.toString(conf.isSyncEnable()))));
+ conf.setSyncServerPort(Integer
+ .parseInt(properties.getProperty("sync_server_port",
+ Integer.toString(conf.getSyncServerPort())).trim()));
conf.setUpdate_historical_data_possibility(Boolean.parseBoolean(
properties.getProperty("update_historical_data_possibility",
- Boolean.toString(conf.isPostbackEnable()))));
+ Boolean.toString(conf.isSyncEnable()))));
conf.setIpWhiteList(properties.getProperty("IP_white_list", conf.getIpWhiteList()));
if (conf.getMemThresholdWarning() <= 0) {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
index be775a5..651e668 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
@@ -77,6 +77,7 @@ import org.apache.iotdb.db.monitor.StatMonitor;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.factory.SeriesReaderFactory;
import org.apache.iotdb.db.query.reader.IReader;
+import org.apache.iotdb.db.sync.conf.Constans;
import org.apache.iotdb.db.utils.MemUtils;
import org.apache.iotdb.db.utils.QueryUtils;
import org.apache.iotdb.db.utils.TimeValuePair;
@@ -724,8 +725,8 @@ public class FileNodeProcessor extends Processor implements IStatistic {
newMultiPassLock.readLock().unlock();
newMultiPassTokenSet.remove(token);
LOGGER.debug("Remove multi token:{}, nspath:{}, new set:{}, lock:{}", token,
- getProcessorName(),
- newMultiPassTokenSet, newMultiPassLock);
+ getProcessorName(),
+ newMultiPassTokenSet, newMultiPassLock);
return true;
} else if (oldMultiPassTokenSet != null && oldMultiPassTokenSet.contains(token)) {
// remove token first, then unlock
@@ -884,8 +885,10 @@ public class FileNodeProcessor extends Processor implements IStatistic {
tsFileResource.getEndTime(entry.getKey()) >= entry.getValue()
&& tsFileResource.getStartTime(entry.getKey()) <= appendFile
.getEndTime(entry.getKey())) {
- String relativeFilePath = "sync" + File.separator + uuid + File.separator + "backup"
- + File.separator + tsFileResource.getRelativePath();
+ String relativeFilePath =
+ Constans.SYNC_SERVER + File.separatorChar + uuid + File.separatorChar
+ + Constans.BACK_UP_DIRECTORY_NAME
+ + File.separatorChar + tsFileResource.getRelativePath();
File newFile = new File(
Directories.getInstance().getTsFileFolder(tsFileResource.getBaseDirIndex()),
relativeFilePath);
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
index cc2581a..739bbbf 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
@@ -34,6 +34,8 @@ public class Constans {
public static final String LAST_LOCAL_FILE_NAME = "last_local_files.txt";
public static final String DATA_SNAPSHOT_NAME = "data-snapshot";
+ public static final String BACK_UP_DIRECTORY_NAME = "backup";
+
/**
* Split data file , block size at each transmission
**/
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
index fe33f2e..05b4659 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
@@ -27,7 +27,7 @@ import org.apache.iotdb.db.metadata.MetadataConstant;
*/
public class SyncSenderConfig {
- private String[] iotdbBufferwriteDirectory = IoTDBDescriptor.getInstance().getConfig()
+ private String[] bufferwriteDirectory = IoTDBDescriptor.getInstance().getConfig()
.getBufferWriteDirs();
private String dataDirectory = IoTDBDescriptor.getInstance().getConfig().getDataDir();
private String uuidPath;
@@ -54,27 +54,27 @@ public class SyncSenderConfig {
uuidPath = dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.UUID_FILE_NAME;
lastFileInfo =
dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME;
- snapshotPaths = new String[iotdbBufferwriteDirectory.length];
- for (int i = 0; i < iotdbBufferwriteDirectory.length; i++) {
- iotdbBufferwriteDirectory[i] = new File(iotdbBufferwriteDirectory[i]).getAbsolutePath();
- if (iotdbBufferwriteDirectory[i].length() > 0
- && iotdbBufferwriteDirectory[i].charAt(iotdbBufferwriteDirectory[i].length() - 1)
+ snapshotPaths = new String[bufferwriteDirectory.length];
+ for (int i = 0; i < bufferwriteDirectory.length; i++) {
+ bufferwriteDirectory[i] = new File(bufferwriteDirectory[i]).getAbsolutePath();
+ if (bufferwriteDirectory[i].length() > 0
+ && bufferwriteDirectory[i].charAt(bufferwriteDirectory[i].length() - 1)
!= File.separatorChar) {
- iotdbBufferwriteDirectory[i] = iotdbBufferwriteDirectory[i] + File.separatorChar;
+ bufferwriteDirectory[i] = bufferwriteDirectory[i] + File.separatorChar;
}
- snapshotPaths[i] = iotdbBufferwriteDirectory[i] + Constans.SYNC_CLIENT + File.separatorChar
+ snapshotPaths[i] = bufferwriteDirectory[i] + Constans.SYNC_CLIENT + File.separatorChar
+ Constans.DATA_SNAPSHOT_NAME
+ File.separatorChar;
}
}
- public String[] getIotdbBufferwriteDirectory() {
- return iotdbBufferwriteDirectory;
+ public String[] getBufferwriteDirectory() {
+ return bufferwriteDirectory;
}
- public void setIotdbBufferwriteDirectory(String[] iotdbBufferwriteDirectory) {
- this.iotdbBufferwriteDirectory = iotdbBufferwriteDirectory;
+ public void setBufferwriteDirectory(String[] bufferwriteDirectory) {
+ this.bufferwriteDirectory = bufferwriteDirectory;
}
public String getDataDirectory() {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
index c04336f..9ca52c1 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
@@ -107,19 +107,18 @@ public class SyncSenderDescriptor {
dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.UUID_FILE_NAME);
conf.setLastFileInfo(
dataDirectory + Constans.SYNC_CLIENT + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME);
- String[] iotdbBufferwriteDirectory = conf.getIotdbBufferwriteDirectory();
- String[] snapshots = new String[conf.getIotdbBufferwriteDirectory().length];
- for (int i = 0; i < conf.getIotdbBufferwriteDirectory().length; i++) {
+ String[] iotdbBufferwriteDirectory = conf.getBufferwriteDirectory();
+ String[] snapshots = new String[conf.getBufferwriteDirectory().length];
+ for (int i = 0; i < conf.getBufferwriteDirectory().length; i++) {
if (iotdbBufferwriteDirectory[i].length() > 0
&& iotdbBufferwriteDirectory[i].charAt(iotdbBufferwriteDirectory[i].length() - 1)
!= File.separatorChar) {
iotdbBufferwriteDirectory[i] = iotdbBufferwriteDirectory[i] + File.separatorChar;
}
snapshots[i] = iotdbBufferwriteDirectory[i] + Constans.SYNC_CLIENT + File.separatorChar
- + Constans.DATA_SNAPSHOT_NAME
- + File.separatorChar;
+ + Constans.DATA_SNAPSHOT_NAME + File.separatorChar;
}
- conf.setIotdbBufferwriteDirectory(iotdbBufferwriteDirectory);
+ conf.setBufferwriteDirectory(iotdbBufferwriteDirectory);
conf.setSnapshotPaths(snapshots);
} catch (IOException e) {
LOGGER.warn("Cannot load config file because {}, use default configuration", e);
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerManager.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerManager.java
index a8baf2f..c723993 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerManager.java
@@ -22,6 +22,7 @@ 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.exception.StartupException;
+import org.apache.iotdb.service.sync.thrift.SyncService;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TBinaryProtocol.Factory;
@@ -58,7 +59,7 @@ public class ServerManager {
Factory protocolFactory;
TProcessor processor;
TThreadPoolServer.Args poolArgs;
- if (!conf.isPostbackEnable()) {
+ if (!conf.isSyncEnable()) {
return;
}
try {
@@ -68,9 +69,9 @@ public class ServerManager {
return;
}
conf.setIpWhiteList(conf.getIpWhiteList().replaceAll(" ", ""));
- serverTransport = new TServerSocket(conf.getPostbackServerPort());
+ serverTransport = new TServerSocket(conf.getSyncServerPort());
protocolFactory = new TBinaryProtocol.Factory();
- processor = new ServerService.Processor<>(new ServerServiceImpl());
+ processor = new SyncService.Processor<>(new ServerServiceImpl());
poolArgs = new TThreadPoolServer.Args(serverTransport);
poolArgs.processor(processor);
poolArgs.protocolFactory(protocolFactory);
@@ -80,7 +81,7 @@ public class ServerManager {
Thread syncServerThread = new Thread(syncServerRunnable, ThreadName.SYNC_SERVER.getName());
syncServerThread.start();
} catch (TTransportException e) {
- throw new StartupException("cannot start sync server.", e);
+ throw new StartupException("Cannot start sync server.", e);
}
}
@@ -88,10 +89,10 @@ public class ServerManager {
* close sync receiver's server.
*/
public void closeServer() {
- if (conf.isPostbackEnable() && poolServer != null) {
+ if (conf.isSyncEnable() && poolServer != null) {
poolServer.stop();
serverTransport.close();
- LOGGER.info("stop sync server.");
+ LOGGER.info("Stop sync server.");
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
index 9674767..8d6661d 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
@@ -37,6 +37,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
+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.Directories;
@@ -70,7 +71,6 @@ 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;
@@ -110,27 +110,34 @@ public class ServerServiceImpl implements SyncService.Iface {
private ThreadLocal<Integer> fileNum = new ThreadLocal<>();
/**
- * IoTDB ioTDBConfig
+ * IoTDB config
**/
- private IoTDBConfig ioTDBConfig = IoTDBDescriptor.getInstance().getConfig();
+ private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
/**
* IoTDB data directory
**/
- private String dataPath = ioTDBConfig.getDataDir();
+ private String dataPath = config.getDataDir();
/**
* IoTDB multiple bufferWrite directory
**/
- private String[] bufferWritePaths = ioTDBConfig.getBufferWriteDirs();
+ private String[] bufferWritePaths = config.getBufferWriteDirs();
/**
* The path to store metadata file of sender
*/
private ThreadLocal<String> schemaFromSenderPath = new ThreadLocal<>();
- /** Sync path of server **/
- private String syncPath;
+ /**
+ * Sync folder path of server
+ **/
+ private String syncFolderPath;
+
+ /**
+ * Sync data path of server
+ */
+ private String syncDataPath;
/**
* Init threadLocal variable
@@ -150,31 +157,28 @@ public class ServerServiceImpl implements SyncService.Iface {
* Verify IP address of sender
*/
@Override
- public boolean checkIdentity(String uuid, String ipAddress) throws TException {
+ public boolean checkIdentity(String uuid, String ipAddress) {
+ Thread.currentThread().setName(ThreadName.SYNC_SERVER.getName());
this.uuid.set(uuid);
initPath();
- return SyncUtils.verifyIPSegment(ioTDBConfig.getIpWhiteList(), ipAddress);
+ return SyncUtils.verifyIPSegment(config.getIpWhiteList(), ipAddress);
}
/**
* Init file path and clear data if last sync process failed.
*/
- private void initPath() throws TException {
+ private void initPath() {
if (dataPath.length() > 0 && dataPath.charAt(dataPath.length() - 1) != File.separatorChar) {
dataPath = dataPath + File.separatorChar;
}
- syncPath = dataPath + SYNC_SERVER + File.separator;
+ syncFolderPath = dataPath + SYNC_SERVER + File.separatorChar + this.uuid.get();
+ syncDataPath = syncFolderPath + File.separatorChar + Constans.DATA_SNAPSHOT_NAME;
schemaFromSenderPath
- .set(syncPath + this.uuid.get() + File.separator + MetadataConstant.METADATA_LOG);
- File syncFileDirectory = new File(syncPath,this.uuid.get());
+ .set(syncFolderPath + File.separator + MetadataConstant.METADATA_LOG);
+ File syncFileDirectory = new File(syncFolderPath, this.uuid.get());
if (syncFileDirectory.exists()
&& Objects.requireNonNull(syncFileDirectory.list()).length != 0) {
- try {
- SyncUtils.deleteFile(syncFileDirectory);
- } catch (IOException e) {
- logger.error("Cannot clear useless metadata file.");
- throw new TException(e);
- }
+ SyncUtils.deleteFile(syncFileDirectory);
}
for (String bufferWritePath : bufferWritePaths) {
if (bufferWritePath.length() > 0
@@ -185,12 +189,7 @@ public class ServerServiceImpl implements SyncService.Iface {
File backupDirectory = new File(backupPath, this.uuid.get());
if (backupDirectory.exists() && Objects.requireNonNull(backupDirectory.list()).length != 0) {
/** if does not exist, it means that the last time sync failed, clear data in the uuid directory and receive the data again **/
- try {
- SyncUtils.deleteFile(backupDirectory);
- } catch (IOException e) {
- logger.error("Cannot clear useless backup data file");
- throw new TException(e);
- }
+ SyncUtils.deleteFile(backupDirectory);
}
}
}
@@ -198,8 +197,9 @@ public class ServerServiceImpl implements SyncService.Iface {
/**
* Acquire schema from sender
*
- * @param status: SUCCESS_STATUS or PROCESSING_STATUS. status = SUCCESS_STATUS : finish receiving schema file, start to sync schema.
- * status = SUCCESS_STATUS : the schema file has not received completely.
+ * @param status: SUCCESS_STATUS or PROCESSING_STATUS. status = SUCCESS_STATUS : finish receiving
+ * schema file, start to sync schema. status = SUCCESS_STATUS : the schema file has not received
+ * completely.
*/
@Override
public void syncSchema(ByteBuffer schema, SyncDataStatus status) {
@@ -215,14 +215,14 @@ public class ServerServiceImpl implements SyncService.Iface {
logger.error("Cannot create file {}", file.getPath());
}
} catch (IOException e) {
- logger.error("Cannot make schema file {}.", file.getPath(),e);
+ logger.error("Cannot make schema file {}.", file.getPath(), e);
}
}
try (FileOutputStream fos = new FileOutputStream(file, true);
FileChannel channel = fos.getChannel()) {
channel.write(schema);
} catch (Exception e) {
- logger.error("Cannot write data to file {}.", file.getPath(),e);
+ logger.error("Cannot write data to file {}.", file.getPath(), e);
}
}
@@ -243,7 +243,7 @@ public class ServerServiceImpl implements SyncService.Iface {
logger.error("Cannot read the file {}.",
schemaFromSenderPath.get(), e);
} catch (IOException e) {
- logger.error("Cannot insert schema to IoTDB.", e);
+ //TODO: how to deal with multiple insert schema
} catch (Exception e) {
logger.error("Parse metadata operation failed.", e);
}
@@ -301,8 +301,8 @@ public class ServerServiceImpl implements SyncService.Iface {
/**
* Start receiving tsfile from sender
*
- * @param status status = SUCCESS_STATUS : finish receiving one tsfile status = PROCESSING_STATUS : tsfile has not received
- * completely.
+ * @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,
@@ -310,7 +310,7 @@ public class ServerServiceImpl implements SyncService.Iface {
String md5OfReceiver = "";
StringBuilder filePathBuilder = new StringBuilder();
FileChannel channel;
- /**Recombination File Path**/
+ /** Recombination File Path **/
for (int i = 0; i < filePathSplit.size(); i++) {
if (i == filePathSplit.size() - 1) {
filePathBuilder.append(filePathSplit.get(i));
@@ -319,10 +319,11 @@ public class ServerServiceImpl implements SyncService.Iface {
}
}
String filePath = filePathBuilder.toString();
- if(syncPath.length() > 0 && syncPath.charAt(syncPath.length()-1)!=File.separatorChar){
- syncPath = syncPath+File.separatorChar;
+ if (syncDataPath.length() > 0
+ && syncDataPath.charAt(syncDataPath.length() - 1) != File.separatorChar) {
+ syncDataPath = syncDataPath + File.separatorChar;
}
- filePath = syncPath + uuid.get() + File.separator + filePath;
+ filePath = syncDataPath + filePath;
if (status == SyncDataStatus.PROCESSING_STATUS) { // there are still data stream to add
File file = new File(filePath);
if (!file.getParentFile().exists()) {
@@ -339,7 +340,7 @@ public class ServerServiceImpl implements SyncService.Iface {
channel = fos.getChannel();
channel.write(dataToReceive);
} catch (IOException e) {
- logger.error("cannot write data to file {}",file.getPath(), e);
+ logger.error("cannot write data to file {}", file.getPath(), e);
}
} else { // all data in the same file has received successfully
@@ -370,14 +371,10 @@ public class ServerServiceImpl implements SyncService.Iface {
@Override
- public boolean load() throws TException {
+ public boolean load() {
getFileNodeInfo();
loadData();
- try {
- SyncUtils.deleteFile(new File(syncPath + this.uuid.get()));
- } catch (IOException e) {
- throw new TException(e);
- }
+ SyncUtils.deleteFile(new File(syncDataPath));
for (String bufferWritePath : bufferWritePaths) {
if (bufferWritePath.length() > 0
&& bufferWritePath.charAt(bufferWritePath.length() - 1) != File.separatorChar) {
@@ -386,12 +383,7 @@ public class ServerServiceImpl implements SyncService.Iface {
String backupPath = bufferWritePath + SYNC_SERVER + File.separator;
File backupDirectory = new File(backupPath, this.uuid.get());
if (backupDirectory.exists() && Objects.requireNonNull(backupDirectory.list()).length != 0) {
- try {
- SyncUtils.deleteFile(backupDirectory);
- } catch (IOException e) {
- logger.error("Cannot clear useless backup data file");
- throw new TException(e);
- }
+ SyncUtils.deleteFile(backupDirectory);
}
}
return true;
@@ -401,7 +393,7 @@ public class ServerServiceImpl implements SyncService.Iface {
* Get all tsfiles' info which are sent from sender, it is prepare for merging these data
*/
public void getFileNodeInfo() {
- File dataFileRoot = new File(syncPath,this.uuid.get());
+ File dataFileRoot = new File(syncDataPath);
File[] files = dataFileRoot.listFiles();
int processedNum = 0;
for (File storageGroupPB : files) {
@@ -432,9 +424,9 @@ public class ServerServiceImpl implements SyncService.Iface {
logger.error("Cannot close file stream {}", fileTF.getPath(), e);
}
}
- fileNodeStartTime.get().put(fileTF.getAbsolutePath(), startTimeMap);
- fileNodeEndTime.get().put(fileTF.getAbsolutePath(), endTimeMap);
- filesPath.add(fileTF.getAbsolutePath());
+ fileNodeStartTime.get().put(fileTF.getPath(), startTimeMap);
+ fileNodeEndTime.get().put(fileTF.getPath(), endTimeMap);
+ filesPath.add(fileTF.getPath());
processedNum++;
if (logger.isInfoEnabled()) {
logger.info(String
@@ -452,8 +444,9 @@ public class ServerServiceImpl implements SyncService.Iface {
* possibility of updating historical data.
*/
public void loadData() {
- if(syncPath.length() > 0 && syncPath.charAt(syncPath.length()-1) != File.separatorChar){
- syncPath =syncPath + File.separatorChar;
+ if (syncDataPath.length() > 0
+ && syncDataPath.charAt(syncDataPath.length() - 1) != File.separatorChar) {
+ syncDataPath = syncDataPath + File.separatorChar;
}
int processedNum = 0;
for (String storageGroup : fileNodeMap.get().keySet()) {
@@ -487,7 +480,7 @@ public class ServerServiceImpl implements SyncService.Iface {
Map<String, Long> endTimeMap = fileNodeEndTime.get().get(path);
// create a new fileNode
- String header = syncPath + uuid.get() + File.separator;
+ String header = syncDataPath;
String relativePath = path.substring(header.length());
TsFileResource fileNode = new TsFileResource(startTimeMap, endTimeMap,
OverflowChangeType.NO_CHANGE,
@@ -496,7 +489,7 @@ public class ServerServiceImpl implements SyncService.Iface {
try {
if (!fileNodeManager.appendFileToFileNode(storageGroup, fileNode, path)) {
// it is a file with overflow data
- if (ioTDBConfig.isUpdate_historical_data_possibility()) {
+ if (config.isUpdate_historical_data_possibility()) {
loadOldData(path);
} else {
List<String> overlapFiles = fileNodeManager.getOverlapFilesFromFileNode(
@@ -735,6 +728,7 @@ public class ServerServiceImpl implements SyncService.Iface {
fileNodeStartTime.remove();
fileNodeEndTime.remove();
schemaFromSenderPath.remove();
+ SyncUtils.deleteFile(new File(syncFolderPath));
logger.info("Synchronization has finished!");
}
@@ -745,4 +739,5 @@ public class ServerServiceImpl implements SyncService.Iface {
public void setFileNodeMap(Map<String, List<String>> fileNodeMap) {
this.fileNodeMap.set(fileNodeMap);
}
+
}
\ No newline at end of file
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
index d2d4258..220745c 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
@@ -97,7 +97,7 @@ public class FileManager {
}
}
}
- LOGGER.info("acquire list of valid files.");
+ LOGGER.info("Acquire list of valid files.");
for (Entry<String, Set<String>> entry : validAllFiles.entrySet()) {
for (String path : entry.getValue()) {
LOGGER.info(path);
@@ -117,10 +117,10 @@ public class FileManager {
if (!file.exists()) {
try {
if (!file.createNewFile()) {
- LOGGER.error("cannot create file {}", file.getAbsoluteFile());
+ LOGGER.error("Cannot create file {}", file.getPath());
}
} catch (IOException e) {
- throw new IOException("cannot get last local file list", e);
+ throw new IOException("Cannot get last local file list", e);
}
} else {
try (BufferedReader bf = new BufferedReader(new FileReader(file))) {
@@ -129,7 +129,7 @@ public class FileManager {
fileList.add(fileName);
}
} catch (IOException e) {
- LOGGER.error("cannot get last local file list when reading file {}.",
+ LOGGER.error("Cannot get last local file list when reading file {}.",
syncConfig.getLastFileInfo());
throw new IOException(e);
}
@@ -160,7 +160,7 @@ public class FileManager {
for (File file : files) {
if (!file.getPath().endsWith(RESTORE_SUFFIX) && !new File(
file.getPath() + RESTORE_SUFFIX).exists()) {
- currentLocalFiles.get(storageGroup.getName()).add(file.getAbsolutePath());
+ currentLocalFiles.get(storageGroup.getName()).add(file.getPath());
}
}
}
@@ -181,7 +181,7 @@ public class FileManager {
}
}
} catch (IOException e) {
- LOGGER.error("cannot back up now local file info", e);
+ LOGGER.error("Cannot back up current local file info", e);
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
index c099c56..0279033 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
@@ -194,12 +194,7 @@ public class FileSenderImpl implements FileSender {
for (String snapshotPath : config.getSnapshotPaths()) {
if (new File(snapshotPath).exists() && new File(snapshotPath).list().length != 0) {
/** It means that the last task of sync does not succeed! Clear the files and start to sync again **/
- try {
- SyncUtils.deleteFile(new File(snapshotPath));
- } catch (IOException e) {
- LOGGER.error("can not delete file {}", snapshotPath);
- throw new IOException(e);
- }
+ SyncUtils.deleteFile(new File(snapshotPath));
}
}
@@ -236,11 +231,7 @@ public class FileSenderImpl implements FileSender {
// 7. clear snapshot
for (String snapshotPath : config.getSnapshotPaths()) {
- try {
- SyncUtils.deleteFile(new File(snapshotPath));
- } catch (IOException e) {
- LOGGER.error("can not delete snapshot", e);
- }
+ SyncUtils.deleteFile(new File(snapshotPath));
}
// 8. notify receiver that synchronization finish
@@ -248,10 +239,10 @@ public class FileSenderImpl implements FileSender {
try {
serviceClient.cleanUp();
} catch (TException e) {
- LOGGER.error("unable to connect to receiver ", e);
+ LOGGER.error("Unable to connect to receiver.", e);
}
transport.close();
- LOGGER.info("sync process has finished");
+ LOGGER.info("Sync process has finished.");
syncStatus = false;
}
@@ -263,21 +254,21 @@ public class FileSenderImpl implements FileSender {
if (validSnapshot.isEmpty()) {
continue;
}
- LOGGER.info("sync process starts to transfer data of storage group {}", entry.getKey());
+ LOGGER.info("Sync process starts to transfer data of storage group {}", entry.getKey());
try {
serviceClient.init(entry.getKey());
} catch (TException e) {
- throw new SyncConnectionException("unable to connect to receiver", e);
+ throw new SyncConnectionException("Unable to connect to receiver", e);
}
syncData(validSnapshot);
if (afterSynchronization()) {
currentLocalFiles.get(entry.getKey()).addAll(validFiles);
fileManager.setCurrentLocalFiles(currentLocalFiles);
fileManager.backupNowLocalFileInfo(config.getLastFileInfo());
- LOGGER.info("sync process has finished storage group {}.", entry.getKey());
+ LOGGER.info("Sync process has finished storage group {}.", entry.getKey());
} else {
throw new SyncConnectionException(
- "receiver cannot sync data, abandon this synchronization");
+ "Receiver cannot sync data, abandon this synchronization");
}
}
}
@@ -297,7 +288,7 @@ public class FileSenderImpl implements FileSender {
transport.open();
} catch (TTransportException e) {
syncStatus = false;
- LOGGER.error("cannot connect to server");
+ LOGGER.error("Cannot connect to server");
throw new SyncConnectionException(e);
}
}
@@ -313,20 +304,18 @@ public class FileSenderImpl implements FileSender {
}
if (!file.exists()) {
try (FileOutputStream out = new FileOutputStream(file)) {
- if (!file.createNewFile()) {
- LOGGER.error("cannot create file {}", file.getPath());
- }
+ file.createNewFile();
uuid = generateUUID();
out.write(uuid.getBytes());
} catch (IOException e) {
- LOGGER.error("cannot write UUID to file {}", file.getPath());
+ LOGGER.error("Cannot write UUID to file {}", file.getPath());
throw new IOException(e);
}
} else {
try (BufferedReader bf = new BufferedReader((new FileReader(uuidPath)))) {
uuid = bf.readLine();
} catch (IOException e) {
- LOGGER.error("cannot read UUID from file{}", file.getPath());
+ LOGGER.error("Cannot read UUID from file{}", file.getPath());
throw new IOException(e);
}
}
@@ -335,7 +324,7 @@ public class FileSenderImpl implements FileSender {
legalConnection = serviceClient.checkIdentity(uuid,
InetAddress.getLocalHost().getHostAddress());
} catch (Exception e) {
- LOGGER.error("cannot confirm identity with receiver");
+ LOGGER.error("Cannot confirm identity with receiver");
throw new SyncConnectionException(e);
}
return legalConnection;
@@ -364,7 +353,7 @@ public class FileSenderImpl implements FileSender {
Files.createLink(link, target);
}
} catch (IOException e) {
- LOGGER.error("can not make fileSnapshot");
+ LOGGER.error("Can not make fileSnapshot");
throw new IOException(e);
}
return validFilesSnapshot;
@@ -421,7 +410,7 @@ public class FileSenderImpl implements FileSender {
String md5OfReceiver = serviceClient.syncData(md5OfSender, filePathSplit,
null, SyncDataStatus.SUCCESS_STATUS);
if (md5OfSender.equals(md5OfReceiver)) {
- LOGGER.info("receiver has received {} successfully.", snapshotFilePath);
+ LOGGER.info("Receiver has received {} successfully.", snapshotFilePath);
break;
}
}
@@ -431,7 +420,7 @@ public class FileSenderImpl implements FileSender {
}
}
} catch (Exception e) {
- throw new SyncConnectionException("cannot sync data with receiver.", e);
+ throw new SyncConnectionException("Cannot sync data with receiver.", e);
}
}
@@ -456,7 +445,7 @@ public class FileSenderImpl implements FileSender {
// 0 represents the schema file has been transferred completely.
serviceClient.syncSchema(null, SyncDataStatus.SUCCESS_STATUS);
} catch (Exception e) {
- LOGGER.error("cannot sync schema ", e);
+ LOGGER.error("Cannot sync schema ", e);
throw new SyncConnectionException(e);
}
}
@@ -468,7 +457,7 @@ public class FileSenderImpl implements FileSender {
successOrNot = serviceClient.load();
} catch (TException e) {
throw new SyncConnectionException(
- "can not finish sync process because sync receiver has broken down.", e);
+ "Can not finish sync process because sync receiver has broken down.", e);
}
return successOrNot;
}
@@ -481,7 +470,7 @@ public class FileSenderImpl implements FileSender {
try {
Socket socket = new Socket("localhost", config.getClientPort());
socket.close();
- LOGGER.error("The sync client has been started!");
+ LOGGER.error("Sync client has been started!");
System.exit(0);
} catch (IOException e) {
try (ServerSocket listenerSocket = new ServerSocket(config.getClientPort())) {
@@ -490,14 +479,13 @@ public class FileSenderImpl implements FileSender {
try {
listenerSocket.accept();
} catch (IOException e2) {
- LOGGER.error("IoTDB sync sender: unable to listen to port{}",
- config.getClientPort(), e2);
+ LOGGER.error("Unable to listen to port{}", config.getClientPort(), e2);
}
}
});
listener.start();
} catch (IOException e1) {
- LOGGER.error("unable to listen to port{}", config.getClientPort());
+ LOGGER.error("Unable to listen to port{}", config.getClientPort());
throw new IOException();
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/utils/SyncUtils.java b/iotdb/src/main/java/org/apache/iotdb/db/utils/SyncUtils.java
index 998aff2..a57d282 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/utils/SyncUtils.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/utils/SyncUtils.java
@@ -1,21 +1,24 @@
/**
- * 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.utils;
import java.io.File;
-import java.io.IOException;
import java.text.DecimalFormat;
import java.util.Map;
import java.util.Map.Entry;
@@ -61,7 +64,8 @@ public class SyncUtils {
if (!new File(snapshotPath).exists()) {
new File(snapshotPath).mkdir();
}
- if(snapshotPath.length() > 0 && snapshotPath.charAt(snapshotPath.length()-1)!=File.separatorChar){
+ if (snapshotPath.length() > 0
+ && snapshotPath.charAt(snapshotPath.length() - 1) != File.separatorChar) {
snapshotPath = snapshotPath + File.separatorChar;
}
return snapshotPath + relativeFilePath;
@@ -129,25 +133,20 @@ public class SyncUtils {
*
* @param file folder file
*/
- public static void deleteFile(File file) throws IOException {
+ public static void deleteFile(File file) {
if (!file.exists()) {
return;
}
if (file.isFile() || Objects.requireNonNull(file.list()).length == 0) {
- if (!file.delete()) {
- throw new IOException(
- String.format("cannot delete file : %s", file.getPath()));
- }
+ file.delete();
} else {
File[] files = file.listFiles();
assert files != null;
for (File f : files) {
deleteFile(f);
- if (!f.delete()) {
- throw new IOException(
- String.format("cannot delete file : %s", f.getPath()));
- }
+ f.delete();
}
+ file.delete();
}
}
}
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
index a23bdbc..6e83e03 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
@@ -33,6 +33,9 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * @author Tianan Li
+ */
public class FileManagerTest {
public static final String POST_BACK_DIRECTORY_TEST = Constans.SYNC_CLIENT + File.separator;
@@ -50,7 +53,7 @@ public class FileManagerTest {
file.getParentFile().mkdirs();
}
if (!file.exists() && !file.createNewFile()) {
- LOGGER.error("Can not create new file {}", file.getAbsoluteFile());
+ LOGGER.error("Can not create new file {}", file.getAbsolutePath());
}
file = new File(SENDER_FILE_PATH_TEST);
if (!file.exists()) {
@@ -78,7 +81,7 @@ public class FileManagerTest {
}
@Test // It tests two classes : backupNowLocalFileInfo and getLastLocalFileList
- public void testBackupNowLocalFileInfo() throws IOException {
+ public void testBackupCurrentLocalFileInfo() throws IOException {
Map<String, Set<String>> allFileList = new HashMap<>();
Random r = new Random(0);
@@ -89,14 +92,14 @@ public class FileManagerTest {
}
String rand = String.valueOf(r.nextInt(10000));
String fileName =
- SENDER_FILE_PATH_TEST + File.separator + String.valueOf(i) + File.separator + rand;
+ SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand;
File file = new File(fileName);
allFileList.get(String.valueOf(i)).add(file.getAbsolutePath());
if (!file.getParentFile().exists()) {
file.getParentFile().mkdirs();
}
if (!file.exists() && !file.createNewFile()) {
- LOGGER.error("Can not create new file {}", file.getAbsoluteFile());
+ LOGGER.error("Can not create new file {}", file.getAbsolutePath());
}
}
}
@@ -125,14 +128,14 @@ public class FileManagerTest {
}
String rand = String.valueOf(r.nextInt(10000));
String fileName =
- SENDER_FILE_PATH_TEST + File.separator + String.valueOf(i) + File.separator + rand;
+ SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand;
File file = new File(fileName);
allFileList.get(String.valueOf(i)).add(file.getAbsolutePath());
if (!file.getParentFile().exists()) {
file.getParentFile().mkdirs();
}
if (!file.exists() && !file.createNewFile()) {
- LOGGER.error("Can not create new file {}", file.getAbsoluteFile());
+ LOGGER.error("Can not create new file {}", file.getAbsolutePath());
}
}
}
@@ -163,7 +166,7 @@ public class FileManagerTest {
}
@Test
- public void testGetNowLocalFileList() throws IOException {
+ public void testGetCurrentLocalFileList() throws IOException {
Map<String, Set<String>> allFileList = new HashMap<>();
Map<String, Set<String>> fileList;
@@ -188,7 +191,7 @@ public class FileManagerTest {
file.getParentFile().mkdirs();
}
if (!file.exists() && !file.createNewFile()) {
- LOGGER.error("Can not create new file {}", file.getAbsoluteFile());
+ LOGGER.error("Can not create new file {}", file.getAbsolutePath());
}
}
}
@@ -226,14 +229,14 @@ public class FileManagerTest {
}
String rand = String.valueOf(r.nextInt(10000));
String fileName =
- SENDER_FILE_PATH_TEST + File.separator + String.valueOf(i) + File.separator + rand;
+ SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand;
File file = new File(fileName);
allFileList.get(String.valueOf(i)).add(file.getAbsolutePath());
if (!file.getParentFile().exists()) {
file.getParentFile().mkdirs();
}
if (!file.exists() && !file.createNewFile()) {
- LOGGER.error("Can not create new file {}", file.getAbsoluteFile());
+ LOGGER.error("Can not create new file {}", file.getAbsolutePath());
}
}
}
@@ -256,11 +259,11 @@ public class FileManagerTest {
}
@Test
- public void testGetSendingFileList() throws IOException {
+ public void testGetValidFileList() throws IOException {
Map<String, Set<String>> allFileList;
Map<String, Set<String>> newFileList = new HashMap<>();
Map<String, Set<String>> sendingFileList;
- Set<String> lastlocalList;
+ Set<String> lastLocalList;
// nowSendingList is empty
@@ -268,9 +271,9 @@ public class FileManagerTest {
manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST});
allFileList = manager.getCurrentLocalFiles();
manager.getLastLocalFileList(LAST_FILE_INFO_TEST);
- lastlocalList = manager.getLastLocalFiles();
+ lastLocalList = manager.getLastLocalFiles();
manager.getValidFileList();
- assert (lastlocalList.isEmpty());
+ assert (lastLocalList.isEmpty());
assert (isEmpty(allFileList));
// add some files
@@ -287,7 +290,7 @@ public class FileManagerTest {
}
String rand = String.valueOf(r.nextInt(10000));
String fileName =
- SENDER_FILE_PATH_TEST + File.separator + String.valueOf(i) + File.separator + rand;
+ SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand;
File file = new File(fileName);
allFileList.get(String.valueOf(i)).add(file.getAbsolutePath());
newFileList.get(String.valueOf(i)).add(file.getAbsolutePath());
@@ -295,7 +298,7 @@ public class FileManagerTest {
file.getParentFile().mkdirs();
}
if (!file.exists() && !file.createNewFile()) {
- LOGGER.error("Can not create new file {}", file.getAbsoluteFile());
+ LOGGER.error("Can not create new file {}", file.getAbsolutePath());
}
}
}
@@ -349,7 +352,7 @@ public class FileManagerTest {
file.getParentFile().mkdirs();
}
if (!file.exists() && !file.createNewFile()) {
- LOGGER.error("Can not create new file {}", file.getAbsoluteFile());
+ LOGGER.error("Can not create new file {}", file.getAbsolutePath());
}
}
}
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientPostBackTest.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientSyncTest.java
similarity index 97%
rename from iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientPostBackTest.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientSyncTest.java
index c988471..c334016 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientPostBackTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientSyncTest.java
@@ -38,17 +38,17 @@ import org.apache.iotdb.jdbc.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class MultipleClientPostBackTest {
+public class MultipleClientSyncTest {
Map<String, ArrayList<String>> timeseriesList = new HashMap();
Map<String, ArrayList<String>> timeseriesList1 = new HashMap();
- private static final Logger LOGGER = LoggerFactory.getLogger(MultipleClientPostBackTest.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(MultipleClientSyncTest.class);
private Set<String> dataSender = new HashSet<>();
private Set<String> dataReceiver = new HashSet<>();
public static void main(String[] args) throws IOException {
- MultipleClientPostBackTest multipleClientPostBackTest = new MultipleClientPostBackTest();
- multipleClientPostBackTest.testPostback();
+ MultipleClientSyncTest multipleClientSyncTest = new MultipleClientSyncTest();
+ multipleClientSyncTest.testPostback();
}
public void testPostback() throws IOException {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/IoTDBSingleClientPostBackTest.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
similarity index 94%
rename from iotdb/src/test/java/org/apache/iotdb/db/sync/sender/IoTDBSingleClientPostBackTest.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
index ccdafff..c5b8c3a 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/IoTDBSingleClientPostBackTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java
@@ -34,23 +34,24 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.SyncConnectionException;
import org.apache.iotdb.db.integration.Constant;
+import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.sync.conf.Constans;
import org.apache.iotdb.db.sync.conf.SyncSenderConfig;
import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
-import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.jdbc.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * The test is to run a complete sync function Before you run the test, make sure receiver has
- * been cleaned up and inited.
+ * The test is to run a complete sync function Before you run the test, make sure receiver has been
+ * cleaned up and inited.
*/
-public class IoTDBSingleClientPostBackTest {
+public class SingleClientSyncTest {
FileSenderImpl fileSenderImpl = FileSenderImpl.getInstance();
private IoTDBConfig conf = IoTDBDescriptor.getInstance().getConfig();
- private String serverIpTest = "192.168.130.17";
+ private String serverIpTest = "192.168.130.7";
private SyncSenderConfig config = SyncSenderDescriptor.getInstance().getConfig();
private Set<String> dataSender = new HashSet<>();
private Set<String> dataReceiver = new HashSet<>();
@@ -145,11 +146,11 @@ public class IoTDBSingleClientPostBackTest {
"insert into root.test.d0(timestamp,s1) values(3000,'1309')",
"insert into root.test.d1.g0(timestamp,s0) values(400,1050)", "merge", "flush",};
private boolean testFlag = Constant.testFlag;
- private static final String POSTBACK = "sync";
- private static final Logger logger = LoggerFactory.getLogger(IoTDBSingleClientPostBackTest.class);
+ private static final String SYNC_CLIENT = Constans.SYNC_CLIENT;
+ private static final Logger logger = LoggerFactory.getLogger(SingleClientSyncTest.class);
public static void main(String[] args) throws Exception {
- IoTDBSingleClientPostBackTest singleClientPostBackTest = new IoTDBSingleClientPostBackTest();
+ SingleClientSyncTest singleClientPostBackTest = new SingleClientSyncTest();
singleClientPostBackTest.setUp();
singleClientPostBackTest.testPostback();
singleClientPostBackTest.tearDown();
@@ -157,21 +158,23 @@ public class IoTDBSingleClientPostBackTest {
}
public void setConfig() {
- config.setUuidPath(config.getDataDirectory() + POSTBACK + File.separator + "uuid.txt");
+ config.setUuidPath(
+ config.getDataDirectory() + SYNC_CLIENT + File.separator + Constans.UUID_FILE_NAME);
config.setLastFileInfo(
- config.getDataDirectory() + POSTBACK + File.separator + "lastLocalFileList.txt");
- String[] iotdbBufferwriteDirectory = config.getIotdbBufferwriteDirectory();
- String[] snapshots = new String[config.getIotdbBufferwriteDirectory().length];
- for (int i = 0; i < config.getIotdbBufferwriteDirectory().length; i++) {
+ config.getDataDirectory() + SYNC_CLIENT + File.separator + Constans.LAST_LOCAL_FILE_NAME);
+ String[] iotdbBufferwriteDirectory = config.getBufferwriteDirectory();
+ String[] snapshots = new String[config.getBufferwriteDirectory().length];
+ for (int i = 0; i < config.getBufferwriteDirectory().length; i++) {
iotdbBufferwriteDirectory[i] = new File(iotdbBufferwriteDirectory[i]).getAbsolutePath();
if (!iotdbBufferwriteDirectory[i].endsWith(File.separator)) {
iotdbBufferwriteDirectory[i] = iotdbBufferwriteDirectory[i] + File.separator;
}
- snapshots[i] = iotdbBufferwriteDirectory[i] + POSTBACK + File.separator + "dataSnapshot"
- + File.separator;
+ snapshots[i] =
+ iotdbBufferwriteDirectory[i] + SYNC_CLIENT + File.separator + Constans.DATA_SNAPSHOT_NAME
+ + File.separator;
}
config.setSnapshotPaths(snapshots);
- config.setIotdbBufferwriteDirectory(iotdbBufferwriteDirectory);
+ config.setBufferwriteDirectory(iotdbBufferwriteDirectory);
config.setServerIp(serverIpTest);
fileSenderImpl.setConfig(config);
}
@@ -204,7 +207,7 @@ public class IoTDBSingleClientPostBackTest {
public void testPostback() throws IOException, SyncConnectionException {
if (testFlag) {
// the first time to sync
- logger.debug("It's the first time to post back!");
+ logger.debug("It's the first time to sync!");
try {
Thread.sleep(2000);
Class.forName(Config.JDBC_DRIVER_NAME);
@@ -271,7 +274,8 @@ public class IoTDBSingleClientPostBackTest {
Connection connection = null;
try {
connection = DriverManager
- .getConnection("jdbc:iotdb://192.168.130.17:6667/", "root", "root");
+ .getConnection(String.format("jdbc:iotdb://%s:6667/", serverIpTest), "root",
+ "root");
Statement statement = connection.createStatement();
boolean hasResultSet = statement.execute("select * from root.vehicle");
if (hasResultSet) {
@@ -298,7 +302,7 @@ public class IoTDBSingleClientPostBackTest {
}
statement.close();
} catch (Exception e) {
- logger.error("",e);
+ logger.error("", e);
} finally {
if (connection != null) {
connection.close();
@@ -318,7 +322,7 @@ public class IoTDBSingleClientPostBackTest {
}
// the second time to sync
- logger.debug("It's the second time to post back!");
+ logger.debug("It's the second time to sync!");
try {
Thread.sleep(2000);
Class.forName(Config.JDBC_DRIVER_NAME);
@@ -400,7 +404,8 @@ public class IoTDBSingleClientPostBackTest {
Connection connection = null;
try {
connection = DriverManager
- .getConnection("jdbc:iotdb://192.168.130.17:6667/", "root", "root");
+ .getConnection(String.format("jdbc:iotdb://%s:6667/", serverIpTest), "root",
+ "root");
Statement statement = connection.createStatement();
boolean hasResultSet = statement.execute("select * from root.vehicle");
if (hasResultSet) {
@@ -446,7 +451,7 @@ public class IoTDBSingleClientPostBackTest {
}
// the third time to sync
- logger.debug("It's the third time to post back!");
+ logger.debug("It's the third time to sync!");
try {
Thread.sleep(2000);
Class.forName(Config.JDBC_DRIVER_NAME);
@@ -475,7 +480,7 @@ public class IoTDBSingleClientPostBackTest {
Thread.sleep(2000);
Class.forName(Config.JDBC_DRIVER_NAME);
try (Connection connection = DriverManager
- .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
+ .getConnection(String.format("jdbc:iotdb://%s:6667/", serverIpTest), "root",
"root")) {
Statement statement = connection.createStatement();
boolean hasResultSet = statement.execute("select * from root.vehicle");
@@ -538,7 +543,7 @@ public class IoTDBSingleClientPostBackTest {
Connection connection = null;
try {
connection = DriverManager
- .getConnection("jdbc:iotdb://192.168.130.17:6667/", "root", "root");
+ .getConnection("jdbc:iotdb://192.168.130.8:6667/", "root", "root");
Statement statement = connection.createStatement();
boolean hasResultSet = statement.execute("select * from root.vehicle");
if (hasResultSet) {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender1.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient1.java
similarity index 97%
rename from iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender1.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient1.java
index 42f1965..35f2f1f 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender1.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient1.java
@@ -37,12 +37,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * CreateDataSender1 is used to generate data of whole timeseries (simulating jilian scene) to test stability of
+ * SyncTestClient1 is used to generate data of whole timeseries (simulating jilian scene) to test stability of
* sync function.
*
* @author Tianan Li
*/
-public class CreateDataSender1 {
+public class SyncTestClient1 {
private static final int TIME_INTERVAL = 0;
private static final int TOTAL_DATA = 2000000;
@@ -58,7 +58,7 @@ public class CreateDataSender1 {
private static final int MAX_FLOAT = 30;
private static final int STRING_LENGTH = 5;
private static final int BATCH_SQL = 10000;
- private static final Logger logger = LoggerFactory.getLogger(CreateDataSender1.class);
+ private static final Logger logger = LoggerFactory.getLogger(SyncTestClient1.class);
/**
* generate time series map from file.
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender2.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient2.java
similarity index 97%
rename from iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender2.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient2.java
index f335e5e..1f0b84c 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender2.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient2.java
@@ -35,12 +35,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * CreateDataSender2 is used to generate data of half timeseries (simulating jilian scene) to test stability of
+ * SyncTestClient2 is used to generate data of half timeseries (simulating jilian scene) to test stability of
* sync function.
*
* @author Tianan Li
*/
-public class CreateDataSender2 {
+public class SyncTestClient2 {
private static final int TIME_INTERVAL = 0;
private static final int TOTAL_DATA = 2000000;
@@ -56,7 +56,7 @@ public class CreateDataSender2 {
private static final int MAX_FLOAT = 30;
private static final int STRING_LENGTH = 5;
private static final int BATCH_SQL = 10000;
- private static final Logger LOGGER = LoggerFactory.getLogger(CreateDataSender2.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(SyncTestClient2.class);
/**
* generate time series map from file.
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender3.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient3.java
similarity index 96%
rename from iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender3.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient3.java
index 83693ca..ed535fb 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender3.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/SyncTestClient3.java
@@ -35,12 +35,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * CreateDataSender3 is used to generate data of another half timeseries (simulating jilian scene) which is
- * different to those in CreateDataSender2 to test stability of sync function.
+ * SyncTestClient3 is used to generate data of another half timeseries (simulating jilian scene) which is
+ * different to those in SyncTestClient2 to test stability of sync function.
*
* @author Tianan Li
*/
-public class CreateDataSender3 {
+public class SyncTestClient3 {
private static final int TIME_INTERVAL = 0;
private static final int TOTAL_DATA = 2000000;
@@ -56,7 +56,7 @@ public class CreateDataSender3 {
private static final int MAX_FLOAT = 30;
private static final int STRING_LENGTH = 5;
private static final int BATCH_SQL = 10000;
- private static final Logger LOGGER = LoggerFactory.getLogger(CreateDataSender3.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(SyncTestClient3.class);
/**
* generate time series map from file.
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/sync/test/Utils.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/Utils.java
index 5f968d4..66322b7 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/sync/test/Utils.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/Utils.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.sync.test;
/**
- * * @author Tianan Li
+ * @author Tianan Li
*/
public class Utils {
diff --git a/iotdb/src/test/resources/start-postBackTest.sh b/iotdb/src/test/resources/start-sync-test.sh
similarity index 100%
rename from iotdb/src/test/resources/start-postBackTest.sh
rename to iotdb/src/test/resources/start-sync-test.sh
diff --git a/iotdb/src/test/resources/stop-postBackTest.sh b/iotdb/src/test/resources/stop-sync-test.sh
similarity index 100%
rename from iotdb/src/test/resources/stop-postBackTest.sh
rename to iotdb/src/test/resources/stop-sync-test.sh
[incubator-iotdb] 02/05: reconstruct thrift sync service
Posted by lt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lta pushed a commit to branch Sync-Reconstruct
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 2f2e1d8e9c0922f9fcd5b4ac73c7e755d53f9865
Author: lta <li...@163.com>
AuthorDate: Sun Mar 17 15:01:26 2019 +0800
reconstruct thrift sync service
---
.../iotdb/db/sync/receiver/ServerServiceImpl.java | 326 +++++++++++----------
.../iotdb/db/sync/sender/FileSenderImpl.java | 22 +-
.../src/main/thrift/sync.thrift | 15 +-
3 files changed, 183 insertions(+), 180 deletions(-)
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
index de2b2a3..435c20b 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
@@ -51,6 +51,8 @@ import org.apache.iotdb.db.metadata.MetadataOperationType;
import org.apache.iotdb.db.qp.executor.OverflowQPExecutor;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.utils.SyncUtils;
+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;
@@ -72,7 +74,7 @@ import org.slf4j.LoggerFactory;
/**
* @author Tianan Li
*/
-public class ServerServiceImpl implements ServerService.Iface {
+public class ServerServiceImpl implements SyncService.Iface {
private static final Logger logger = LoggerFactory.getLogger(ServerServiceImpl.class);
private static final FileNodeManager fileNodeManager = FileNodeManager.getInstance();
@@ -145,75 +147,6 @@ public class ServerServiceImpl implements ServerService.Iface {
return SyncUtils.verifyIPSegment(config.getIpWhiteList(), ipAddress);
}
- /**
- * Start receiving tsfile from sender
- *
- * @param status status = 0 : finish receiving one tsfile status = 1 : a tsfile has not received
- * completely.
- */
- @Override
- public String startReceiving(String md5OfSender, List<String> filePathSplit,
- ByteBuffer dataToReceive, int status) throws TException {
- String md5OfReceiver = "";
- StringBuilder filePathBuilder = new StringBuilder();
- FileChannel channel;
- for (int i = 0; i < filePathSplit.size(); i++) {
- if (i == filePathSplit.size() - 1) {
- filePathBuilder.append(filePathSplit.get(i));
- } else {
- filePathBuilder.append(filePathSplit.get(i)).append(File.separator);
- }
- }
- String filePath = filePathBuilder.toString();
- filePath = postbackPath + uuid.get() + File.separator + filePath;
- if (status == 1) { // there are still data stream to add
- File file = new File(filePath);
- if (!file.getParentFile().exists()) {
- try {
- file.getParentFile().mkdirs();
- if (!file.createNewFile()) {
- logger.error("IoTDB post back receiver: cannot create file {}", file.getAbsoluteFile());
- }
- } catch (IOException e) {
- logger.error("IoTDB post back receiver: cannot make file", e);
- }
- }
- try (FileOutputStream fos = new FileOutputStream(file, true)) {// append new data
- channel = fos.getChannel();
- channel.write(dataToReceive);
- } catch (IOException e) {
- logger.error("IoTDB post back receiver: cannot write data to file", e);
-
- }
- } else { // all data in the same file has received successfully
- try (FileInputStream fis = new FileInputStream(filePath)) {
- MessageDigest md = MessageDigest.getInstance("MD5");
- int mBufferSize = 8 * 1024 * 1024;
- byte[] buffer = new byte[mBufferSize];
- 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);
- if (logger.isInfoEnabled()) {
- logger.info(String
- .format("IoTDB post back receiver : Receiver has received %d files from sender",
- fileNum.get()));
- }
- } else {
- if (!new File(filePath).delete()) {
- logger.error("IoTDB post back receiver : Receiver can not delete file {}",
- new File(filePath).getAbsolutePath());
- }
- }
- } catch (Exception e) {
- logger.error("IoTDB post back receiver: cannot generate md5", e);
- }
- }
- return md5OfReceiver;
- }
/**
* Get schema from sender
@@ -222,8 +155,8 @@ public class ServerServiceImpl implements ServerService.Iface {
* IoTDB through jdbc status = 1 : the schema file has not received completely.
*/
@Override
- public void getSchema(ByteBuffer schema, int status) {
- if (status == 0) {
+ public void getSchema(ByteBuffer schema, SyncDataStatus status) {
+ if (status == SyncDataStatus.SUCCESS_STATUS) {
/** sync metadata, include storage group and timeseries **/
syncMetadata();
} else {
@@ -319,10 +252,81 @@ public class ServerServiceImpl implements ServerService.Iface {
}
}
+ /**
+ * Start receiving tsfile from sender
+ *
+ * @param status status = 0 : finish receiving one tsfile status = 1 : a tsfile has not received
+ * completely.
+ */
+ @Override
+ public String receiveData(String md5OfSender, List<String> filePathSplit,
+ ByteBuffer dataToReceive, SyncDataStatus status) {
+ String md5OfReceiver = "";
+ StringBuilder filePathBuilder = new StringBuilder();
+ FileChannel channel;
+ for (int i = 0; i < filePathSplit.size(); i++) {
+ if (i == filePathSplit.size() - 1) {
+ filePathBuilder.append(filePathSplit.get(i));
+ } else {
+ filePathBuilder.append(filePathSplit.get(i)).append(File.separator);
+ }
+ }
+ String filePath = filePathBuilder.toString();
+ filePath = postbackPath + uuid.get() + File.separator + 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();
+ if (!file.createNewFile()) {
+ logger.error("IoTDB post back receiver: cannot create file {}", file.getAbsoluteFile());
+ }
+ } catch (IOException e) {
+ logger.error("IoTDB post back receiver: cannot make file", e);
+ }
+ }
+ try (FileOutputStream fos = new FileOutputStream(file, true)) {// append new data
+ channel = fos.getChannel();
+ channel.write(dataToReceive);
+ } catch (IOException e) {
+ logger.error("IoTDB post back receiver: cannot write data to file", e);
+
+ }
+ } else { // all data in the same file has received successfully
+ try (FileInputStream fis = new FileInputStream(filePath)) {
+ MessageDigest md = MessageDigest.getInstance("MD5");
+ int mBufferSize = 8 * 1024 * 1024;
+ byte[] buffer = new byte[mBufferSize];
+ 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);
+ if (logger.isInfoEnabled()) {
+ logger.info(String
+ .format("IoTDB post back receiver : Receiver has received %d files from sender",
+ fileNum.get()));
+ }
+ } else {
+ if (!new File(filePath).delete()) {
+ logger.error("IoTDB post back receiver : Receiver can not delete file {}",
+ new File(filePath).getAbsolutePath());
+ }
+ }
+ } catch (Exception e) {
+ logger.error("IoTDB post back receiver: cannot generate md5", e);
+ }
+ }
+ return md5OfReceiver;
+ }
+
+
@Override
- public boolean merge() throws TException {
+ public boolean load() throws TException {
getFileNodeInfo();
- mergeData();
+ loadData();
try {
SyncUtils.deleteFile(new File(postbackPath + this.uuid.get()));
} catch (IOException e) {
@@ -345,24 +349,9 @@ public class ServerServiceImpl implements ServerService.Iface {
}
/**
- * Release threadLocal variable resources
- */
- @Override
- public void afterReceiving() {
- uuid.remove();
- fileNum.remove();
- fileNodeMap.remove();
- fileNodeStartTime.remove();
- fileNodeEndTime.remove();
- schemaFromSenderPath.remove();
- logger.info("IoTDB post back receiver: the postBack has finished!");
- }
-
- /**
* Get all tsfiles' info which are sent from sender, it is prepare for merging these data
*/
- @Override
- public void getFileNodeInfo() throws TException {
+ public void getFileNodeInfo() {
String filePath = postbackPath + uuid.get() + File.separator + "data";
File root = new File(filePath);
File[] files = root.listFiles();
@@ -411,11 +400,84 @@ public class ServerServiceImpl implements ServerService.Iface {
}
}
+
+ /**
+ * It is to merge data. If data in the tsfile is new, append the tsfile to the storage group
+ * directly. If data in the tsfile is old, it has two strategy to merge.It depends on the
+ * possibility of updating historical data.
+ */
+ public void loadData() throws TException {
+ int num = 0;
+ for (String storageGroup : fileNodeMap.get().keySet()) {
+ List<String> filesPath = fileNodeMap.get().get(storageGroup);
+ // before load extern tsFile, it is necessary to order files in the same SG
+ for (int i = 0; i < filesPath.size(); i++) {
+ for (int j = i + 1; j < filesPath.size(); j++) {
+ boolean swapOrNot = false;
+ Map<String, Long> startTimeI = fileNodeStartTime.get().get(filesPath.get(i));
+ Map<String, Long> endTimeI = fileNodeStartTime.get().get(filesPath.get(i));
+ Map<String, Long> startTimeJ = fileNodeStartTime.get().get(filesPath.get(j));
+ Map<String, Long> endTimeJ = fileNodeStartTime.get().get(filesPath.get(j));
+ for (String deviceId : endTimeI.keySet()) {
+ if (startTimeJ.containsKey(deviceId) && startTimeI.get(deviceId) >
+ endTimeJ.get(deviceId)) {
+ swapOrNot = true;
+ break;
+ }
+ }
+ if (swapOrNot) {
+ String temp = filesPath.get(i);
+ filesPath.set(i, filesPath.get(j));
+ filesPath.set(j, temp);
+ }
+ }
+ }
+
+ for (String path : filesPath) {
+ // get startTimeMap and endTimeMap
+ Map<String, Long> startTimeMap = fileNodeStartTime.get().get(path);
+ Map<String, Long> endTimeMap = fileNodeEndTime.get().get(path);
+
+ // create a new fileNode
+ String header = postbackPath + uuid.get() + File.separator + "data" + File.separator;
+ String relativePath = path.substring(header.length());
+ TsFileResource fileNode = new TsFileResource(startTimeMap, endTimeMap,
+ OverflowChangeType.NO_CHANGE,
+ Directories.getInstance().getNextFolderIndexForTsFile(), relativePath);
+ // call interface of load external file
+ try {
+ if (!fileNodeManager.appendFileToFileNode(storageGroup, fileNode, path)) {
+ // it is a file with overflow data
+ if (config.isUpdate_historical_data_possibility()) {
+ loadOldData(path);
+ } else {
+ List<String> overlapFiles = fileNodeManager.getOverlapFilesFromFileNode(
+ storageGroup,
+ fileNode, uuid.get());
+ if (overlapFiles.isEmpty()) {
+ loadOldData(path);
+ } else {
+ loadOldData(path, overlapFiles);
+ }
+ }
+ }
+ } catch (FileNodeManagerException e) {
+ logger.error("IoTDB receiver : Can not load external file ", e);
+ }
+
+ num++;
+ if (logger.isInfoEnabled()) {
+ logger.info(String
+ .format("IoTDB receiver : Merging files has completed : %d/%d", num, fileNum.get()));
+ }
+ }
+ }
+ }
+
/**
* Insert all data in the tsfile into IoTDB.
*/
- @Override
- public void mergeOldData(String filePath) throws TException {
+ public void loadOldData(String filePath) throws TException {
Set<String> timeseriesSet = new HashSet<>();
TsFileSequenceReader reader = null;
OverflowQPExecutor insertExecutor = new OverflowQPExecutor();
@@ -493,7 +555,7 @@ public class ServerServiceImpl implements ServerService.Iface {
*
* @param overlapFiles:files which are conflict with the sync file
*/
- public void mergeOldData(String filePath, List<String> overlapFiles) {
+ public void loadOldData(String filePath, List<String> overlapFiles) {
Set<String> timeseriesList = new HashSet<>();
TsFileSequenceReader reader = null;
OverflowQPExecutor insertExecutor = new OverflowQPExecutor();
@@ -615,77 +677,17 @@ public class ServerServiceImpl implements ServerService.Iface {
}
/**
- * It is to merge data. If data in the tsfile is new, append the tsfile to the storage group
- * directly. If data in the tsfile is old, it has two strategy to merge.It depends on the
- * possibility of updating historical data.
+ * Release threadLocal variable resources
*/
@Override
- public void mergeData() throws TException {
- int num = 0;
- for (String storageGroup : fileNodeMap.get().keySet()) {
- List<String> filesPath = fileNodeMap.get().get(storageGroup);
- // before load extern tsFile, it is necessary to order files in the same SG
- for (int i = 0; i < filesPath.size(); i++) {
- for (int j = i + 1; j < filesPath.size(); j++) {
- boolean swapOrNot = false;
- Map<String, Long> startTimeI = fileNodeStartTime.get().get(filesPath.get(i));
- Map<String, Long> endTimeI = fileNodeStartTime.get().get(filesPath.get(i));
- Map<String, Long> startTimeJ = fileNodeStartTime.get().get(filesPath.get(j));
- Map<String, Long> endTimeJ = fileNodeStartTime.get().get(filesPath.get(j));
- for (String deviceId : endTimeI.keySet()) {
- if (startTimeJ.containsKey(deviceId) && startTimeI.get(deviceId) >
- endTimeJ.get(deviceId)) {
- swapOrNot = true;
- break;
- }
- }
- if (swapOrNot) {
- String temp = filesPath.get(i);
- filesPath.set(i, filesPath.get(j));
- filesPath.set(j, temp);
- }
- }
- }
-
- for (String path : filesPath) {
- // get startTimeMap and endTimeMap
- Map<String, Long> startTimeMap = fileNodeStartTime.get().get(path);
- Map<String, Long> endTimeMap = fileNodeEndTime.get().get(path);
-
- // create a new fileNode
- String header = postbackPath + uuid.get() + File.separator + "data" + File.separator;
- String relativePath = path.substring(header.length());
- TsFileResource fileNode = new TsFileResource(startTimeMap, endTimeMap,
- OverflowChangeType.NO_CHANGE,
- Directories.getInstance().getNextFolderIndexForTsFile(), relativePath);
- // call interface of load external file
- try {
- if (!fileNodeManager.appendFileToFileNode(storageGroup, fileNode, path)) {
- // it is a file with overflow data
- if (config.isUpdate_historical_data_possibility()) {
- mergeOldData(path);
- } else {
- List<String> overlapFiles = fileNodeManager.getOverlapFilesFromFileNode(
- storageGroup,
- fileNode, uuid.get());
- if (overlapFiles.isEmpty()) {
- mergeOldData(path);
- } else {
- mergeOldData(path, overlapFiles);
- }
- }
- }
- } catch (FileNodeManagerException e) {
- logger.error("IoTDB receiver : Can not load external file ", e);
- }
-
- num++;
- if (logger.isInfoEnabled()) {
- logger.info(String
- .format("IoTDB receiver : Merging files has completed : %d/%d", num, fileNum.get()));
- }
- }
- }
+ public void cleanUp() {
+ uuid.remove();
+ fileNum.remove();
+ fileNodeMap.remove();
+ fileNodeStartTime.remove();
+ fileNodeEndTime.remove();
+ schemaFromSenderPath.remove();
+ logger.info("IoTDB post back receiver: the postBack has finished!");
}
public Map<String, List<String>> getFileNodeMap() {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
index ea34a29..e8939de 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
@@ -48,8 +48,9 @@ import org.apache.iotdb.db.exception.SyncConnectionException;
import org.apache.iotdb.db.sync.conf.Constans;
import org.apache.iotdb.db.sync.conf.SyncSenderConfig;
import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
-import org.apache.iotdb.db.sync.receiver.ServerService;
import org.apache.iotdb.db.utils.SyncUtils;
+import org.apache.iotdb.service.sync.thrift.SyncDataStatus;
+import org.apache.iotdb.service.sync.thrift.SyncService;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
@@ -68,7 +69,7 @@ public class FileSenderImpl implements FileSender {
private static final Logger LOGGER = LoggerFactory.getLogger(FileSenderImpl.class);
private TTransport transport;
- private ServerService.Client serviceClient;
+ private SyncService.Client serviceClient;
private List<String> schema = new ArrayList<>();
/**
@@ -245,7 +246,7 @@ public class FileSenderImpl implements FileSender {
// 8. notify receiver that synchronization finish
// At this point the synchronization has finished even if connection fails
try {
- serviceClient.afterReceiving();
+ serviceClient.cleanUp();
} catch (TException e) {
LOGGER.error("unable to connect to receiver ", e);
}
@@ -291,7 +292,7 @@ public class FileSenderImpl implements FileSender {
public void establishConnection(String serverIp, int serverPort) throws SyncConnectionException {
transport = new TSocket(serverIp, serverPort);
TProtocol protocol = new TBinaryProtocol(transport);
- serviceClient = new ServerService.Client(protocol);
+ serviceClient = new SyncService.Client(protocol);
try {
transport.open();
} catch (TTransportException e) {
@@ -401,7 +402,8 @@ public class FileSenderImpl implements FileSender {
bos.write(buffer, 0, data);
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
- serviceClient.startReceiving(null, filePathSplit, buffToSend, 1);
+ serviceClient
+ .receiveData(null, filePathSplit, buffToSend, SyncDataStatus.PROCESSING_STATUS);
}
bos.close();
}
@@ -416,8 +418,8 @@ public class FileSenderImpl implements FileSender {
// the file is sent successfully
String md5OfSender = (new BigInteger(1, md.digest())).toString(16);
- String md5OfReceiver = serviceClient.startReceiving(md5OfSender, filePathSplit,
- null, 0);
+ String md5OfReceiver = serviceClient.receiveData(md5OfSender, filePathSplit,
+ null, SyncDataStatus.SUCCESS_STATUS);
if (md5OfSender.equals(md5OfReceiver)) {
LOGGER.info("receiver has received {} successfully.", snapshotFilePath);
break;
@@ -448,11 +450,11 @@ public class FileSenderImpl implements FileSender {
ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
bos.reset();
// 1 represents there is still schema buffer to send.
- serviceClient.getSchema(buffToSend, 1);
+ serviceClient.getSchema(buffToSend, SyncDataStatus.PROCESSING_STATUS);
}
bos.close();
// 0 represents the schema file has been transferred completely.
- serviceClient.getSchema(null, 0);
+ serviceClient.getSchema(null, SyncDataStatus.SUCCESS_STATUS);
} catch (Exception e) {
LOGGER.error("cannot sync schema ", e);
throw new SyncConnectionException(e);
@@ -463,7 +465,7 @@ public class FileSenderImpl implements FileSender {
public boolean afterSynchronization() throws SyncConnectionException {
boolean successOrNot;
try {
- successOrNot = serviceClient.merge();
+ successOrNot = serviceClient.load();
} catch (TException e) {
throw new SyncConnectionException(
"can not finish sync process because sync receiver has broken down.", e);
diff --git a/iotdb/src/main/thrift/SyncServerService.thrift b/service-rpc/src/main/thrift/sync.thrift
similarity index 80%
rename from iotdb/src/main/thrift/SyncServerService.thrift
rename to service-rpc/src/main/thrift/sync.thrift
index d633d25..a947507 100755
--- a/iotdb/src/main/thrift/SyncServerService.thrift
+++ b/service-rpc/src/main/thrift/sync.thrift
@@ -16,23 +16,22 @@
* specific language governing permissions and limitations
* under the License.
*/
-namespace java org.apache.iotdb.db.sync.receiver
+namespace java org.apache.iotdb.service.sync.thrift
typedef i32 int
typedef i16 short
typedef i64 long
-enum SYNC_STATUS {
+enum SyncDataStatus {
SUCCESS_STATUS,
- SYNC_STATUS
-
+ PROCESSING_STATUS
}
-service ServerService{
+service SyncService{
bool getUUID(1:string uuid, 2:string address)
- void getSchema(1:binary buff, 2:int status)
- string receiveData(1:string md5, 2:list<string> filename, 3:binary buff, 4:int status)
+ void getSchema(1:binary buff, 2:SyncDataStatus status)
+ string receiveData(1:string md5, 2:list<string> filename, 3:binary buff, 4:SyncDataStatus status)
bool load()
- void afterReceiving()
+ void cleanUp()
void init(1:string storageGroup)
}
\ No newline at end of file
[incubator-iotdb] 01/05: rename postback to sync and reconstruct
sync client
Posted by lt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lta pushed a commit to branch Sync-Reconstruct
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 96f371022fb2bbe5cbf80f9f6e5df927344b19e5
Author: lta <li...@163.com>
AuthorDate: Sun Mar 17 14:36:03 2019 +0800
rename postback to sync and reconstruct sync client
---
...rt-postBackClient.bat => start-sync-client.bat} | 0
...tart-postBackClient.sh => start-sync-client.sh} | 0
...top-postBackClient.bat => stop-sync-client.bat} | 0
...{stop-postBackClient.sh => stop-sync-client.sh} | 0
iotdb/iotdb/conf/iotdb-engine.properties | 2 +-
...ent.properties => iotdb-sync-client.properties} | 0
.../org/apache/iotdb/db/concurrent/ThreadName.java | 5 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 4 +-
.../db/engine/filenode/FileNodeProcessor.java | 2 +-
.../db/exception/SyncConnectionException.java} | 75 +--
.../iotdb/db/postback/sender/FileSenderImpl.java | 510 --------------------
.../java/org/apache/iotdb/db/service/IoTDB.java | 2 +-
.../org/apache/iotdb/db/sync/conf/Constans.java} | 75 +--
.../conf/SyncSenderConfig.java} | 46 +-
.../conf/SyncSenderDescriptor.java} | 77 +--
.../{postback => sync}/receiver/ServerManager.java | 26 +-
.../receiver/ServerServiceImpl.java | 439 +++++++++---------
.../db/{postback => sync}/sender/FileManager.java | 87 ++--
.../db/{postback => sync}/sender/FileSender.java | 29 +-
.../iotdb/db/sync/sender/FileSenderImpl.java | 516 +++++++++++++++++++++
.../utils/{PostbackUtils.java => SyncUtils.java} | 74 ++-
...rverService.thrift => SyncServerService.thrift} | 16 +-
.../filenodev2/FileNodeManagerBenchmark.java | 2 +-
.../{postback => sync}/sender/FileManagerTest.java | 21 +-
.../sender/IoTDBSingleClientPostBackTest.java | 28 +-
.../sender/MultipleClientPostBackTest.java | 2 +-
.../iotdb/db/sync/test}/CreateDataSender1.java | 10 +-
.../iotdb/db/sync/test}/CreateDataSender2.java | 8 +-
.../iotdb/db/sync/test}/CreateDataSender3.java | 8 +-
.../org/apache/iotdb/db/sync/test}/RandomNum.java | 2 +-
.../java/org/apache/iotdb/db/sync/test}/Utils.java | 4 +-
31 files changed, 1078 insertions(+), 992 deletions(-)
diff --git a/iotdb/iotdb/bin/start-postBackClient.bat b/iotdb/iotdb/bin/start-sync-client.bat
similarity index 100%
rename from iotdb/iotdb/bin/start-postBackClient.bat
rename to iotdb/iotdb/bin/start-sync-client.bat
diff --git a/iotdb/iotdb/bin/start-postBackClient.sh b/iotdb/iotdb/bin/start-sync-client.sh
similarity index 100%
rename from iotdb/iotdb/bin/start-postBackClient.sh
rename to iotdb/iotdb/bin/start-sync-client.sh
diff --git a/iotdb/iotdb/bin/stop-postBackClient.bat b/iotdb/iotdb/bin/stop-sync-client.bat
similarity index 100%
rename from iotdb/iotdb/bin/stop-postBackClient.bat
rename to iotdb/iotdb/bin/stop-sync-client.bat
diff --git a/iotdb/iotdb/bin/stop-postBackClient.sh b/iotdb/iotdb/bin/stop-sync-client.sh
similarity index 100%
rename from iotdb/iotdb/bin/stop-postBackClient.sh
rename to iotdb/iotdb/bin/stop-sync-client.sh
diff --git a/iotdb/iotdb/conf/iotdb-engine.properties b/iotdb/iotdb/conf/iotdb-engine.properties
index fe6d154..cdca52c 100644
--- a/iotdb/iotdb/conf/iotdb-engine.properties
+++ b/iotdb/iotdb/conf/iotdb-engine.properties
@@ -211,7 +211,7 @@ postback_server_port=5555
# White IP list of Postback client.
# Please use the form of network segment to present the range of IP, for example: 192.168.0.0/16
# If there are more than one IP segment, please separate them by commas
-# The default is to allow all IP to postback
+# The default is to allow all IP to sync
IP_white_list=0.0.0.0/0
# Choose a postBack strategy of merging historical data:
diff --git a/iotdb/iotdb/conf/iotdb-postbackClient.properties b/iotdb/iotdb/conf/iotdb-sync-client.properties
similarity index 100%
rename from iotdb/iotdb/conf/iotdb-postbackClient.properties
rename to iotdb/iotdb/conf/iotdb-sync-client.properties
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java b/iotdb/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
index 04ca847..6053da6 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/concurrent/ThreadName.java
@@ -34,7 +34,10 @@ public enum ThreadName {
FLUSH_SERVICE("Flush-ServerServiceImpl"),
WAL_DAEMON("IoTDB-MultiFileLogNodeManager-Sync-Thread"),
WAL_FORCE_DAEMON("IoTDB-MultiFileLogNodeManager-Force-Thread"),
- INDEX_SERVICE("Index-ServerServiceImpl");
+ INDEX_SERVICE("Index-ServerServiceImpl"),
+ SYNC_CLIENT("Sync-Client"),
+ SYNC_SERVER("Sync-Server"),
+ SYNC_MONITOR("Sync-Monitor");
private String name;
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 22f17ae..815e36b 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -253,11 +253,11 @@ public class IoTDBConfig {
*/
private int maxLogEntrySize = 4 * 1024 * 1024;
/**
- * Is this IoTDB instance a receiver of postback or not.
+ * Is this IoTDB instance a receiver of sync or not.
*/
private boolean isPostbackEnable = true;
/**
- * If this IoTDB instance is a receiver of postback, set the server port.
+ * If this IoTDB instance is a receiver of sync, set the server port.
*/
private int postbackServerPort = 5555;
/*
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
index 6f56ebc..3fddd8d 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
@@ -886,7 +886,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
tsFileResource.getEndTime(entry.getKey()) >= entry.getValue()
&& tsFileResource.getStartTime(entry.getKey()) <= appendFile
.getEndTime(entry.getKey())) {
- String relativeFilePath = "postback" + File.separator + uuid + File.separator + "backup"
+ String relativeFilePath = "sync" + File.separator + uuid + File.separator + "backup"
+ File.separator + tsFileResource.getRelativePath();
File newFile = new File(
Directories.getInstance().getTsFileFolder(tsFileResource.getBaseDirIndex()),
diff --git a/iotdb/src/main/thrift/ServerService.thrift b/iotdb/src/main/java/org/apache/iotdb/db/exception/SyncConnectionException.java
old mode 100755
new mode 100644
similarity index 61%
copy from iotdb/src/main/thrift/ServerService.thrift
copy to iotdb/src/main/java/org/apache/iotdb/db/exception/SyncConnectionException.java
index cdcc47c..5b04e36
--- a/iotdb/src/main/thrift/ServerService.thrift
+++ b/iotdb/src/main/java/org/apache/iotdb/db/exception/SyncConnectionException.java
@@ -1,34 +1,41 @@
-/*
- * 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.
- */
- namespace java org.apache.iotdb.db.postback.receiver
-
-typedef i32 int
-typedef i16 short
-typedef i64 long
-service ServerService{
- bool getUUID(1:string uuid, 2:string address)
- string startReceiving(1:string md5, 2:list<string> filename, 3:binary buff, 4:int status)
- void getFileNodeInfo()
- void mergeOldData(1:string path)
- void mergeData()
- void getSchema(1:binary buff, 2:int status)
- bool merge()
- void afterReceiving()
- void init(1:string storageGroup)
-}
\ No newline at end of file
+/**
+ * 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.exception;
+
+/**
+ * @author Tianan Li
+ */
+public class SyncConnectionException extends Exception {
+
+
+ private static final long serialVersionUID = -6661904365503849681L;
+
+ public SyncConnectionException(String message) {
+ super(message);
+ }
+
+ public SyncConnectionException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public SyncConnectionException(Throwable cause) {
+ super(cause);
+ }
+
+}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileSenderImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileSenderImpl.java
deleted file mode 100644
index 0b38d38..0000000
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileSenderImpl.java
+++ /dev/null
@@ -1,510 +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.postback.sender;
-
-import java.io.BufferedReader;
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.FileReader;
-import java.io.IOException;
-import java.math.BigInteger;
-import java.net.InetAddress;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-import java.nio.file.FileSystems;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.security.MessageDigest;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.UUID;
-import org.apache.iotdb.db.postback.conf.PostBackSenderConfig;
-import org.apache.iotdb.db.postback.conf.PostBackSenderDescriptor;
-import org.apache.iotdb.db.postback.receiver.ServerService;
-import org.apache.iotdb.db.utils.PostbackUtils;
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * The class is to transfer tsfiles that needs to postback to receiver.
- *
- * @author lta
- */
-public class FileSenderImpl implements FileSender {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(FileSenderImpl.class);
- private TTransport transport;
- private ServerService.Client clientOfServer;
- private List<String> schema = new ArrayList<>();
- private String uuid;// Mark the identity of sender
- /**
- * Mark whether connection of sender and receiver has broken down or not.
- */
- private boolean connectionOrElse;
- private PostBackSenderConfig config = PostBackSenderDescriptor.getInstance().getConfig();
- private Date lastPostBackTime = new Date(); // Mark the start time of last postback
- private boolean postBackStatus = false; // If true, postback is in execution.
- private static final String POST_BACK_SENDER_POSTBACK_PROCESS_HAS_FAILED = "IoTDB post back sender : postback process has failed!";
- private Map<String, Set<String>> sendingFileSnapshotList = new HashMap<>();
-
- private FileSenderImpl() {
- }
-
- public static final FileSenderImpl getInstance() {
- return TransferHolder.INSTANCE;
- }
-
- /**
- * Create a sender and send files to the receiver.
- *
- * @param args not used
- */
- public static void main(String[] args) throws InterruptedException {
- FileSenderImpl fileSenderImpl = new FileSenderImpl();
- fileSenderImpl.verifyPort();
- Thread monitor = new Thread(fileSenderImpl::monitorPostbackStatus);
- monitor.start();
- fileSenderImpl.timedTask();
- }
-
- public void setConfig(PostBackSenderConfig config) {
- this.config = config;
- }
-
- private void getConnection(String serverIP, int serverPort) {
- connectToReceiver(serverIP, serverPort);
- if (connectionOrElse && !transferUUID(config.getUuidPath())) {
- LOGGER.error(
- "IoTDB post back sender: Sorry! You do not have the permission to "
- + "connect to postback receiver!");
- connectionOrElse = false;
-
- }
- }
-
- /**
- * Establish a connection between the sender and the receiver.
- *
- * @param serverIp the ip address of the receiver
- * @param serverPort must be same with port receiver set.
- */
- @Override
- public void connectToReceiver(String serverIp, int serverPort) {
- transport = new TSocket(serverIp, serverPort);
- TProtocol protocol = new TBinaryProtocol(transport);
- clientOfServer = new ServerService.Client(protocol);
- try {
- transport.open();
- } catch (TTransportException e) {
- LOGGER.error("IoTDB post back sender: cannot connect to server", e);
- connectionOrElse = false;
- }
- }
-
- /**
- * UUID marks the identity of sender for receiver.
- */
- @Override
- public boolean transferUUID(String uuidPath) {
- File file = new File(uuidPath);
- if (!file.getParentFile().exists()) {
- file.getParentFile().mkdirs();
- }
- if (!file.exists()) {
- try (FileOutputStream out = new FileOutputStream(file)) {
- if (!file.createNewFile()) {
- LOGGER.error("IoTDB post back sender: cannot create file {}",
- file.getAbsoluteFile());
- }
- uuid = "PB" + UUID.randomUUID().toString().replaceAll("-", "");
- out.write(uuid.getBytes());
- } catch (Exception e) {
- LOGGER.error("IoTDB post back sender: cannot write UUID to file", e);
- connectionOrElse = false;
- }
- } else {
- try (BufferedReader bf = new BufferedReader((new FileReader(uuidPath)))) {
- uuid = bf.readLine();
- } catch (IOException e) {
- LOGGER.error("IoTDB post back sender: cannot read UUID from file", e);
- connectionOrElse = false;
- }
- }
- boolean legalConnectionOrNot = true;
- try {
- legalConnectionOrNot = clientOfServer.getUUID(uuid,
- InetAddress.getLocalHost().getHostAddress());
- } catch (TException e) {
- LOGGER.error("IoTDB post back sender: cannot send UUID to receiver", e);
- connectionOrElse = false;
- } catch (UnknownHostException e) {
- LOGGER.error("IoTDB post back sender: unable to get local host", e);
- legalConnectionOrNot = false;
- }
- return legalConnectionOrNot;
- }
-
- /**
- * Create snapshots for those sending files.
- */
- @Override
- public Set<String> makeFileSnapshot(Set<String> sendingFileList) {
- Set<String> sendingSnapshotFileList = new HashSet<>();
- try {
- for (String filePath : sendingFileList) {
- String snapshotFilePath = PostbackUtils.getSnapshotFilePath(filePath);
- sendingSnapshotFileList.add(snapshotFilePath);
- File newFile = new File(snapshotFilePath);
- if (!newFile.getParentFile().exists()) {
- newFile.getParentFile().mkdirs();
- }
- Path link = FileSystems.getDefault().getPath(snapshotFilePath);
- Path target = FileSystems.getDefault().getPath(filePath);
- Files.createLink(link, target);
- }
- } catch (IOException e) {
- LOGGER.error("IoTDB post back sender: can not make fileSnapshot", e);
- }
- return sendingSnapshotFileList;
- }
-
- /**
- * Transfer data of a storage group to receiver.
- *
- * @param fileSnapshotList list of sending snapshot files in a storage group.
- */
- @Override
- public void transferData(Set<String> fileSnapshotList) {
- try {
- int num = 0;
- for (String snapshotFilePath : fileSnapshotList) {
- num++;
- File file = new File(snapshotFilePath);
- List<String> filePathSplit = new ArrayList<>();
- String os = System.getProperty("os.name");
- if (os.toLowerCase().startsWith("windows")) {
- String[] name = snapshotFilePath.split(File.separator + File.separator);
- filePathSplit.add("data");
- filePathSplit.add(name[name.length - 2]);
- filePathSplit.add(name[name.length - 1]);
- } else {
- String[] name = snapshotFilePath.split(File.separator);
- filePathSplit.add("data");
- filePathSplit.add(name[name.length - 2]);
- filePathSplit.add(name[name.length - 1]);
- }
- while (true) {
- // Send all data to receiver
- try (FileInputStream fis = new FileInputStream(file)) {
- int mBufferSize = 64 * 1024 * 1024;
- ByteArrayOutputStream bos = new ByteArrayOutputStream(mBufferSize);
- byte[] buffer = new byte[mBufferSize];
- int n;
- while ((n = fis.read(buffer)) != -1) { // cut the file into pieces to send
- bos.write(buffer, 0, n);
- ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
- bos.reset();
- clientOfServer.startReceiving(null, filePathSplit, buffToSend, 1);
- }
- bos.close();
- }
-
- // Get md5 of the file.
- MessageDigest md = MessageDigest.getInstance("MD5");
- try (FileInputStream fis = new FileInputStream(file)) {
- int mBufferSize = 8 * 1024 * 1024;
- byte[] buffer = new byte[mBufferSize];
- int m;
- while ((m = fis.read(buffer)) != -1) {
- md.update(buffer, 0, m);
- }
- }
-
- // the file is sent successfully
- String md5OfSender = (new BigInteger(1, md.digest())).toString(16);
- String md5OfReceiver = clientOfServer.startReceiving(md5OfSender, filePathSplit,
- null, 0);
- if (md5OfSender.equals(md5OfReceiver)) {
- LOGGER.info("IoTDB sender: receiver has received {} successfully.", snapshotFilePath);
- break;
- }
- }
- LOGGER.info(String
- .format("IoTDB sender : Task of sending files to receiver has completed %d/%d.", num,
- fileSnapshotList.size()));
- }
- } catch (TException e) {
- LOGGER.error("IoTDB post back sender: cannot sending data because receiver has broken down.");
- connectionOrElse = false;
- } catch (Exception e) {
- LOGGER.error("IoTDB post back sender: cannot sending data ", e);
- connectionOrElse = false;
- }
- }
-
- /**
- * Sending schema to receiver.
- *
- * @param schemaPath the path of the schema file.
- */
- @Override
- public void sendSchema(String schemaPath) {
- try (FileInputStream fis = new FileInputStream(new File(schemaPath))) {
- int mBufferSize = 4 * 1024 * 1024;
- ByteArrayOutputStream bos = new ByteArrayOutputStream(mBufferSize);
- byte[] buffer = new byte[mBufferSize];
- int n;
- while ((n = fis.read(buffer)) != -1) { // cut the file into pieces to send
- bos.write(buffer, 0, n);
- ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
- bos.reset();
- // 1 represents there is still schema buffer to send.
- clientOfServer.getSchema(buffToSend, 1);
- }
- bos.close();
- // 0 represents the schema file has been transferred completely.
- clientOfServer.getSchema(null, 0);
- } catch (Exception e) {
- LOGGER.error("IoTDB post back sender : cannot send schema from mlog.txt ", e);
- connectionOrElse = false;
- }
- }
-
- @Override
- public boolean afterSending() {
- boolean successOrNot = false;
- try {
- successOrNot = clientOfServer.merge();
- } catch (TException e) {
- LOGGER.error(
- "IoTDB post back sender : can not finish postback process because postback "
- + "receiver has broken down.");
- transport.close();
- }
- return successOrNot;
- }
-
- public List<String> getSchema() {
- return schema;
- }
-
- /**
- * The method is to verify whether the client port is bind or not, ensuring that only one client
- * is running.
- */
- private void verifyPort() {
- try {
- Socket socket = new Socket("localhost", config.getClientPort());
- socket.close();
- LOGGER.error("The postback client has been started!");
- System.exit(0);
- } catch (IOException e) {
- try {
- ServerSocket listenerSocket = new ServerSocket(config.getClientPort());
- Thread listener = new Thread(() -> {
- while (true) {
- try {
- listenerSocket.accept();
- } catch (IOException e2) {
- LOGGER.error("IoTDB post back sender: unable to listen to port{}",
- config.getClientPort(), e2);
- }
- }
- });
- listener.start();
- } catch (IOException e1) {
- LOGGER.error("IoTDB post back sender: unable to listen to port{}",
- config.getClientPort(), e1);
- }
- }
- }
-
- /**
- * Monitor postback status.
- */
- private void monitorPostbackStatus() {
- Date oldTime = new Date();
- while (true) {
- Date currentTime = new Date();
- if (currentTime.getTime() / 1000 == oldTime.getTime() / 1000) {
- continue;
- }
- if ((currentTime.getTime() - lastPostBackTime.getTime())
- % (config.getUploadCycleInSeconds() * 1000) == 0) {
- oldTime = currentTime;
- if (postBackStatus) {
- LOGGER.info("IoTDB post back sender : postback process is in execution!");
- }
- }
- }
- }
-
- /**
- * Start postback task in a certain time.
- */
- public void timedTask() throws InterruptedException {
- postback();
- lastPostBackTime = new Date();
- Date currentTime;
- while (true) {
- Thread.sleep(2000);
- currentTime = new Date();
- if (currentTime.getTime() - lastPostBackTime.getTime()
- > config.getUploadCycleInSeconds() * 1000) {
- lastPostBackTime = currentTime;
- postback();
- }
- }
- }
-
- /**
- * Execute a postback task.
- */
- @Override
- public void postback() {
-
- for (String snapshotPath : config.getSnapshotPaths()) {
- if (new File(snapshotPath).exists() && new File(snapshotPath).list().length != 0) {
- // it means that the last task of postback does not succeed! Clear the files and
- // start to postback again
- try {
- PostbackUtils.deleteFile(new File(snapshotPath));
- } catch (IOException e) {
- LOGGER.error("can not delete file {}", snapshotPath, e);
- }
- }
- }
-
- postBackStatus = true;
- connectionOrElse = true;
-
- // connect to postback server
- getConnection(config.getServerIp(), config.getServerPort());
- if (!connectionOrElse) {
- LOGGER.info(POST_BACK_SENDER_POSTBACK_PROCESS_HAS_FAILED);
- postBackStatus = false;
- return;
- }
-
- FileManager fileManager = FileManager.getInstance();
- fileManager.init();
- Map<String, Set<String>> sendingFileList = fileManager.getSendingFiles();
- Map<String, Set<String>> nowLocalFileList = fileManager.getCurrentLocalFiles();
- if (PostbackUtils.isEmpty(sendingFileList)) {
- LOGGER.info("IoTDB post back sender : there has no file to postback !");
- postBackStatus = false;
- return;
- }
-
- // create snapshot
- for (Entry<String, Set<String>> entry : sendingFileList.entrySet()) {
- sendingFileSnapshotList.put(entry.getKey(), makeFileSnapshot(entry.getValue()));
- }
-
- sendSchema(config.getSchemaPath());
- if (!connectionOrElse) {
- transport.close();
- LOGGER.info(POST_BACK_SENDER_POSTBACK_PROCESS_HAS_FAILED);
- postBackStatus = false;
- return;
- }
- for (Entry<String, Set<String>> entry : sendingFileList.entrySet()) {
- Set<String> sendingList = entry.getValue();
- Set<String> sendingSnapshotList = sendingFileSnapshotList.get(entry.getKey());
- if (sendingSnapshotList.isEmpty()) {
- continue;
- }
- LOGGER.info("IoTDB post back sender : postback process starts to transfer data of "
- + "storage group {}.", entry.getKey());
- try {
- clientOfServer.init(entry.getKey());
- } catch (TException e) {
- connectionOrElse = false;
- LOGGER.error("IoTDB post back sender : unable to connect to receiver", e);
- }
- if (!connectionOrElse) {
- transport.close();
- LOGGER.info(POST_BACK_SENDER_POSTBACK_PROCESS_HAS_FAILED);
- postBackStatus = false;
- return;
- }
- transferData(sendingSnapshotList);
- if (!connectionOrElse) {
- transport.close();
- LOGGER.info(POST_BACK_SENDER_POSTBACK_PROCESS_HAS_FAILED);
- postBackStatus = false;
- return;
- }
- if (afterSending()) {
- nowLocalFileList.get(entry.getKey()).addAll(sendingList);
- fileManager.setCurrentLocalFiles(nowLocalFileList);
- fileManager.backupNowLocalFileInfo(config.getLastFileInfo());
- LOGGER.info("IoTDB post back sender : the postBack has finished storage group {}.",
- entry.getKey());
- } else {
- LOGGER.info(POST_BACK_SENDER_POSTBACK_PROCESS_HAS_FAILED);
- postBackStatus = false;
- return;
- }
- }
- for (String snapshotPath : config.getSnapshotPaths()) {
- try {
- PostbackUtils.deleteFile(new File(snapshotPath));
- } catch (IOException e) {
- LOGGER.error(" ", e);
- }
- }
- try {
- clientOfServer.afterReceiving();
- } catch (TException e) {
- connectionOrElse = false;
- LOGGER.error("IoTDB post back sender : unable to connect to receiver ", e);
- }
- if (!connectionOrElse) {
- transport.close();
- LOGGER.info(POST_BACK_SENDER_POSTBACK_PROCESS_HAS_FAILED);
- postBackStatus = false;
- return;
- }
- transport.close();
- LOGGER.info("IoTDB post back sender : postback process has finished!");
- postBackStatus = false;
- }
-
- private static class TransferHolder {
-
- private static final FileSenderImpl INSTANCE = new FileSenderImpl();
- }
-}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/service/IoTDB.java b/iotdb/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index 9689166..fead066 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -33,7 +33,7 @@ import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.builder.ExceptionBuilder;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.monitor.StatMonitor;
-import org.apache.iotdb.db.postback.receiver.ServerManager;
+import org.apache.iotdb.db.sync.receiver.ServerManager;
import org.apache.iotdb.db.query.control.FileReaderManager;
import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
import org.apache.iotdb.db.writelog.manager.WriteLogNodeManager;
diff --git a/iotdb/src/main/thrift/ServerService.thrift b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
old mode 100755
new mode 100644
similarity index 58%
copy from iotdb/src/main/thrift/ServerService.thrift
copy to iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
index cdcc47c..31e0ad4
--- a/iotdb/src/main/thrift/ServerService.thrift
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/Constans.java
@@ -1,34 +1,41 @@
-/*
- * 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.
- */
- namespace java org.apache.iotdb.db.postback.receiver
-
-typedef i32 int
-typedef i16 short
-typedef i64 long
-service ServerService{
- bool getUUID(1:string uuid, 2:string address)
- string startReceiving(1:string md5, 2:list<string> filename, 3:binary buff, 4:int status)
- void getFileNodeInfo()
- void mergeOldData(1:string path)
- void mergeData()
- void getSchema(1:binary buff, 2:int status)
- bool merge()
- void afterReceiving()
- void init(1:string storageGroup)
-}
\ No newline at end of file
+/**
+ * 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.sync.conf;
+
+/**
+ * @author Tianan Li
+ */
+public class Constans {
+
+ private Constans() {
+ }
+
+ public static final String CONFIG_NAME = "iotdb-sync-client.properties";
+ public static final String SYNC = "sync";
+
+ public static final String UUID_FILE_NAME = "uuid.txt";
+ public static final String LAST_LOCAL_FILE_NAME = "last_local_files.txt";
+ public static final String DATA_SNAPSHOT_NAME = "data_snapshot";
+
+ /**
+ * Split data file , block size at each transmission
+ **/
+ public static final int DATA_CHUNK_SIZE = 64 * 1024 * 1024;
+
+}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/conf/PostBackSenderConfig.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
similarity index 64%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/conf/PostBackSenderConfig.java
rename to iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
index b0d1894..0def77c 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/conf/PostBackSenderConfig.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderConfig.java
@@ -16,35 +16,59 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.conf;
+package org.apache.iotdb.db.sync.conf;
import java.io.File;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.metadata.MetadataConstant;
/**
- * @author lta
+ * @author Tianan Li
*/
-public class PostBackSenderConfig {
-
- public static final String CONFIG_NAME = "iotdb-postbackClient.properties";
+public class SyncSenderConfig {
private String[] iotdbBufferwriteDirectory = IoTDBDescriptor.getInstance().getConfig()
.getBufferWriteDirs();
- private String dataDirectory =
- new File(IoTDBDescriptor.getInstance().getConfig().getDataDir()).getAbsolutePath()
- + File.separator;
+ private String dataDirectory = IoTDBDescriptor.getInstance().getConfig().getDataDir();
private String uuidPath;
private String lastFileInfo;
private String[] snapshotPaths;
- private String schemaPath =
- new File(IoTDBDescriptor.getInstance().getConfig().getMetadataDir()).getAbsolutePath()
- + File.separator + "mlog.txt";
+ private String schemaPath;
private String serverIp = "127.0.0.1";
private int serverPort = 5555;
private int clientPort = 6666;
private int uploadCycleInSeconds = 10;
private boolean clearEnable = false;
+ public void init() {
+ String metadataDirPath = IoTDBDescriptor.getInstance().getConfig().getMetadataDir();
+ if (metadataDirPath.length() > 0
+ && metadataDirPath.charAt(metadataDirPath.length() - 1) != File.separatorChar) {
+ metadataDirPath = metadataDirPath + File.separatorChar;
+ }
+ schemaPath = metadataDirPath + MetadataConstant.METADATA_LOG;
+ if (dataDirectory.length() > 0
+ && dataDirectory.charAt(dataDirectory.length() - 1) != File.separatorChar) {
+ dataDirectory += File.separatorChar;
+ }
+ uuidPath = dataDirectory + Constans.SYNC + File.separatorChar + Constans.UUID_FILE_NAME;
+ lastFileInfo =
+ dataDirectory + Constans.SYNC + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME;
+ snapshotPaths = new String[iotdbBufferwriteDirectory.length];
+ for (int i = 0; i < iotdbBufferwriteDirectory.length; i++) {
+ iotdbBufferwriteDirectory[i] = new File(iotdbBufferwriteDirectory[i]).getAbsolutePath();
+ if (iotdbBufferwriteDirectory[i].length() > 0
+ && iotdbBufferwriteDirectory[i].charAt(iotdbBufferwriteDirectory[i].length() - 1)
+ != File.separatorChar) {
+ iotdbBufferwriteDirectory[i] = iotdbBufferwriteDirectory[i] + File.separatorChar;
+ }
+ snapshotPaths[i] = iotdbBufferwriteDirectory[i] + Constans.SYNC + File.separatorChar
+ + Constans.DATA_SNAPSHOT_NAME
+ + File.separatorChar;
+ }
+
+ }
+
public String[] getIotdbBufferwriteDirectory() {
return iotdbBufferwriteDirectory;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/conf/PostBackSenderDescriptor.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
similarity index 59%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/conf/PostBackSenderDescriptor.java
rename to iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
index d4f5983..04a5e7d 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/conf/PostBackSenderDescriptor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/conf/SyncSenderDescriptor.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.conf;
+package org.apache.iotdb.db.sync.conf;
import java.io.File;
import java.io.FileInputStream;
@@ -29,53 +29,61 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * @author lta
+ * @author Tianan Li
*/
-public class PostBackSenderDescriptor {
+public class SyncSenderDescriptor {
- private static final Logger LOGGER = LoggerFactory.getLogger(PostBackSenderDescriptor.class);
- private static final String POSTBACK = "postback";
- private PostBackSenderConfig conf = new PostBackSenderConfig();
+ private static final Logger LOGGER = LoggerFactory.getLogger(SyncSenderDescriptor.class);
+ private SyncSenderConfig conf = new SyncSenderConfig();
- private PostBackSenderDescriptor() {
+ private SyncSenderDescriptor() {
loadProps();
}
- public static final PostBackSenderDescriptor getInstance() {
+ public static final SyncSenderDescriptor getInstance() {
return PostBackDescriptorHolder.INSTANCE;
}
- public PostBackSenderConfig getConfig() {
+ public SyncSenderConfig getConfig() {
return conf;
}
- public void setConfig(PostBackSenderConfig conf) {
+ public void setConfig(SyncSenderConfig conf) {
this.conf = conf;
}
/**
- * load an properties file and set TsfileDBConfig variables
+ * load an properties file and set sync config variables
*/
private void loadProps() {
+ conf.init();
+ InputStream inputStream;
String url = System.getProperty(IoTDBConstant.IOTDB_CONF, null);
if (url == null) {
url = System.getProperty(IoTDBConstant.IOTDB_HOME, null);
if (url != null) {
url = url + File.separatorChar + "conf" + File.separatorChar
- + PostBackSenderConfig.CONFIG_NAME;
+ + Constans.CONFIG_NAME;
} else {
LOGGER.warn(
"Cannot find IOTDB_HOME or IOTDB_CONF environment variable when loading config file {}, use default configuration",
- PostBackSenderConfig.CONFIG_NAME);
+ Constans.CONFIG_NAME);
return;
}
} else {
- url += (File.separatorChar + PostBackSenderConfig.CONFIG_NAME);
+ url += (File.separatorChar + Constans.CONFIG_NAME);
}
- LOGGER.info("Start to read config file {}", url);
+ try {
+ inputStream = new FileInputStream(new File(url));
+ } catch (FileNotFoundException e) {
+ LOGGER.warn("Fail to find sync config file {}", url, e);
+ return;
+ }
+
+ LOGGER.info("Start to read sync config file {}", url);
Properties properties = new Properties();
- try (InputStream inputStream = new FileInputStream(new File(url))) {
+ try {
properties.load(inputStream);
conf.setServerIp(properties.getProperty("server_ip", conf.getServerIp()));
@@ -88,21 +96,28 @@ public class PostBackSenderDescriptor {
.getProperty("upload_cycle_in_seconds",
Integer.toString(conf.getUploadCycleInSeconds()))));
conf.setSchemaPath(properties.getProperty("iotdb_schema_directory", conf.getSchemaPath()));
- conf.setClearEnable(Boolean
- .parseBoolean(
- properties.getProperty("is_clear_enable", Boolean.toString(conf.getClearEnable()))));
- conf.setUuidPath(conf.getDataDirectory() + POSTBACK + File.separator + "uuid.txt");
+ conf.setDataDirectory(
+ properties.getProperty("iotdb_bufferWrite_directory", conf.getDataDirectory()));
+ String dataDirectory = conf.getDataDirectory();
+ if (dataDirectory.length() > 0
+ && dataDirectory.charAt(dataDirectory.length() - 1) != File.separatorChar) {
+ dataDirectory += File.separatorChar;
+ }
+ conf.setUuidPath(
+ dataDirectory + Constans.SYNC + File.separatorChar + Constans.UUID_FILE_NAME);
conf.setLastFileInfo(
- conf.getDataDirectory() + POSTBACK + File.separator + "lastLocalFileList.txt");
+ dataDirectory + Constans.SYNC + File.separatorChar + Constans.LAST_LOCAL_FILE_NAME);
String[] iotdbBufferwriteDirectory = conf.getIotdbBufferwriteDirectory();
String[] snapshots = new String[conf.getIotdbBufferwriteDirectory().length];
for (int i = 0; i < conf.getIotdbBufferwriteDirectory().length; i++) {
- iotdbBufferwriteDirectory[i] = new File(iotdbBufferwriteDirectory[i]).getAbsolutePath();
- if (!iotdbBufferwriteDirectory[i].endsWith(File.separator)) {
- iotdbBufferwriteDirectory[i] = iotdbBufferwriteDirectory[i] + File.separator;
+ if (iotdbBufferwriteDirectory[i].length() > 0
+ && iotdbBufferwriteDirectory[i].charAt(iotdbBufferwriteDirectory[i].length() - 1)
+ != File.separatorChar) {
+ iotdbBufferwriteDirectory[i] = iotdbBufferwriteDirectory[i] + File.separatorChar;
}
- snapshots[i] = iotdbBufferwriteDirectory[i] + POSTBACK + File.separator + "dataSnapshot"
- + File.separator;
+ snapshots[i] = iotdbBufferwriteDirectory[i] + Constans.SYNC + File.separatorChar
+ + Constans.DATA_SNAPSHOT_NAME
+ + File.separatorChar;
}
conf.setIotdbBufferwriteDirectory(iotdbBufferwriteDirectory);
conf.setSnapshotPaths(snapshots);
@@ -110,11 +125,19 @@ public class PostBackSenderDescriptor {
LOGGER.warn("Cannot load config file because {}, use default configuration", e);
} catch (Exception e) {
LOGGER.warn("Error format in config file because {}, use default configuration", e);
+ } finally {
+ if (inputStream != null) {
+ try {
+ inputStream.close();
+ } catch (IOException e) {
+ LOGGER.error("Fail to close sync config file input stream because ", e);
+ }
+ }
}
}
private static class PostBackDescriptorHolder {
- private static final PostBackSenderDescriptor INSTANCE = new PostBackSenderDescriptor();
+ private static final SyncSenderDescriptor INSTANCE = new SyncSenderDescriptor();
}
}
\ No newline at end of file
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerManager.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerManager.java
similarity index 81%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerManager.java
rename to iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerManager.java
index 0a82d71..a8baf2f 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerManager.java
@@ -16,8 +16,9 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.receiver;
+package org.apache.iotdb.db.sync.receiver;
+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.exception.StartupException;
@@ -32,9 +33,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * receiver server.
+ * sync receiver server.
*
- * @author lta
+ * @author Tianan Li
*/
public class ServerManager {
@@ -51,7 +52,7 @@ public class ServerManager {
}
/**
- * start postback receiver's server.
+ * start sync receiver's server.
*/
public void startServer() throws StartupException {
Factory protocolFactory;
@@ -63,8 +64,7 @@ public class ServerManager {
try {
if (conf.getIpWhiteList() == null) {
LOGGER.error(
- "IoTDB post back receiver: Postback server failed to start because IP white "
- + "list is null, please set IP white list!");
+ "Sync server failed to start because IP white list is null, please set IP white list.");
return;
}
conf.setIpWhiteList(conf.getIpWhiteList().replaceAll(" ", ""));
@@ -75,23 +75,23 @@ public class ServerManager {
poolArgs.processor(processor);
poolArgs.protocolFactory(protocolFactory);
poolServer = new TThreadPoolServer(poolArgs);
- LOGGER.info("Postback server has started.");
- Runnable runnable = () -> poolServer.serve();
- Thread thread = new Thread(runnable);
- thread.start();
+ LOGGER.info("Sync server has started.");
+ Runnable syncServerRunnable = () -> poolServer.serve();
+ Thread syncServerThread = new Thread(syncServerRunnable, ThreadName.SYNC_SERVER.getName());
+ syncServerThread.start();
} catch (TTransportException e) {
- throw new StartupException("IoTDB post back receiver: cannot start postback server.", e);
+ throw new StartupException("cannot start sync server.", e);
}
}
/**
- * close postback receiver's server.
+ * close sync receiver's server.
*/
public void closeServer() {
if (conf.isPostbackEnable() && poolServer != null) {
poolServer.stop();
serverTransport.close();
- LOGGER.info("Stop postback server.");
+ LOGGER.info("stop sync server.");
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerServiceImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
similarity index 65%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerServiceImpl.java
rename to iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
index 7bbdb16..de2b2a3 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/receiver/ServerServiceImpl.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.receiver;
+package org.apache.iotdb.db.sync.receiver;
import java.io.BufferedReader;
import java.io.File;
@@ -28,26 +28,36 @@ import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.security.MessageDigest;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.*;
-
+import java.util.ArrayList;
+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.iotdb.db.conf.IoTDBConfig;
-import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.Directories;
import org.apache.iotdb.db.engine.filenode.FileNodeManager;
-import org.apache.iotdb.db.engine.filenode.TsFileResource;
import org.apache.iotdb.db.engine.filenode.OverflowChangeType;
+import org.apache.iotdb.db.engine.filenode.TsFileResource;
import org.apache.iotdb.db.exception.FileNodeManagerException;
-import org.apache.iotdb.db.utils.PostbackUtils;
+import org.apache.iotdb.db.exception.MetadataArgsErrorException;
+import org.apache.iotdb.db.exception.PathErrorException;
+import org.apache.iotdb.db.exception.ProcessorException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.metadata.MetadataOperationType;
+import org.apache.iotdb.db.qp.executor.OverflowQPExecutor;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.utils.SyncUtils;
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;
@@ -60,14 +70,14 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * @author lta
+ * @author Tianan Li
*/
public class ServerServiceImpl implements ServerService.Iface {
private static final Logger logger = LoggerFactory.getLogger(ServerServiceImpl.class);
private static final FileNodeManager fileNodeManager = FileNodeManager.getInstance();
- private static final String JDBC_DRIVER_NAME = "org.apache.iotdb.jdbc.IoTDBDriver";
- private static final String POSTBACK = "postback";
+ private static final MManager metadataManger = MManager.getInstance();
+ private static final String POSTBACK = "sync";
private ThreadLocal<String> uuid = new ThreadLocal<>();
// String means Storage Group,List means the set of new Files(AbsulutePath) in local IoTDB
// String means AbsulutePath of new Files
@@ -81,7 +91,8 @@ public class ServerServiceImpl implements ServerService.Iface {
private IoTDBConfig tsfileDBconfig = IoTDBDescriptor.getInstance().getConfig();
private String postbackPath;
// Absolute seriesPath of IoTDB data directory
- private String dataPath = new File(tsfileDBconfig.getDataDir()).getAbsolutePath() + File.separator;
+ private String dataPath =
+ new File(tsfileDBconfig.getDataDir()).getAbsolutePath() + File.separator;
// Absolute paths of IoTDB bufferWrite directory
private String[] bufferWritePaths = tsfileDBconfig.getBufferWriteDirs();
private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
@@ -91,9 +102,11 @@ public class ServerServiceImpl implements ServerService.Iface {
*/
@Override
public void init(String storageGroup) {
- logger.info(String.format(
- "IoTDB post back receiver: postback process starts to receive data of storage group {}."),
- storageGroup);
+ if (logger.isInfoEnabled()) {
+ logger.info(
+ "IoTDB post back receiver: sync process starts to receive data of storage group {}",
+ storageGroup);
+ }
fileNum.set(0);
fileNodeMap.set(new HashMap<>());
fileNodeStartTime.set(new HashMap<>());
@@ -111,7 +124,7 @@ public class ServerServiceImpl implements ServerService.Iface {
if (new File(postbackPath + this.uuid.get()).exists()
&& new File(postbackPath + this.uuid.get()).list().length != 0) {
try {
- PostbackUtils.deleteFile(new File(postbackPath + this.uuid.get()));
+ SyncUtils.deleteFile(new File(postbackPath + this.uuid.get()));
} catch (IOException e) {
throw new TException(e);
}
@@ -120,16 +133,16 @@ public class ServerServiceImpl implements ServerService.Iface {
String backupPath = bufferWritePath + POSTBACK + File.separator;
if (new File(backupPath + this.uuid.get()).exists()
&& new File(backupPath + this.uuid.get()).list().length != 0) {
- // if does not exist, it means that the last time postback failed, clear uuid
+ // if does not exist, it means that the last time sync failed, clear uuid
// data and receive the data again
try {
- PostbackUtils.deleteFile(new File(backupPath + this.uuid.get()));
+ SyncUtils.deleteFile(new File(backupPath + this.uuid.get()));
} catch (IOException e) {
throw new TException(e);
}
}
}
- return PostbackUtils.verifyIPSegment(config.getIpWhiteList(), ipAddress);
+ return SyncUtils.verifyIPSegment(config.getIpWhiteList(), ipAddress);
}
/**
@@ -143,12 +156,12 @@ public class ServerServiceImpl implements ServerService.Iface {
ByteBuffer dataToReceive, int status) throws TException {
String md5OfReceiver = "";
StringBuilder filePathBuilder = new StringBuilder();
- FileChannel channel = null;
+ FileChannel channel;
for (int i = 0; i < filePathSplit.size(); i++) {
if (i == filePathSplit.size() - 1) {
- filePathBuilder = filePathBuilder.append(filePathSplit.get(i));
+ filePathBuilder.append(filePathSplit.get(i));
} else {
- filePathBuilder = filePathBuilder.append(filePathSplit.get(i)).append(File.separator);
+ filePathBuilder.append(filePathSplit.get(i)).append(File.separator);
}
}
String filePath = filePathBuilder.toString();
@@ -184,11 +197,16 @@ public class ServerServiceImpl implements ServerService.Iface {
md5OfReceiver = (new BigInteger(1, md.digest())).toString(16);
if (md5OfSender.equals(md5OfReceiver)) {
fileNum.set(fileNum.get() + 1);
- logger.info(String
- .format("IoTDB post back receiver : Receiver has received %d files from sender!",
- fileNum.get()));
+ if (logger.isInfoEnabled()) {
+ logger.info(String
+ .format("IoTDB post back receiver : Receiver has received %d files from sender",
+ fileNum.get()));
+ }
} else {
- new File(filePath).delete();
+ if (!new File(filePath).delete()) {
+ logger.error("IoTDB post back receiver : Receiver can not delete file {}",
+ new File(filePath).getAbsolutePath());
+ }
}
} catch (Exception e) {
logger.error("IoTDB post back receiver: cannot generate md5", e);
@@ -205,58 +223,9 @@ public class ServerServiceImpl implements ServerService.Iface {
*/
@Override
public void getSchema(ByteBuffer schema, int status) {
- FileOutputStream fos;
- FileChannel channel;
if (status == 0) {
- Statement statement = null;
- try (Connection connection = DriverManager.getConnection("jdbc:iotdb://localhost:" +
- config.getRpcPort() + "/", IoTDBConstant.ADMIN_NAME, IoTDBConstant.ADMIN_PW)) {
- Class.forName(JDBC_DRIVER_NAME);
- statement = connection.createStatement();
-
- try (BufferedReader bf = new BufferedReader(
- new java.io.FileReader(schemaFromSenderPath.get()))) {
- String data;
- statement.clearBatch();
- int count = 0;
- while ((data = bf.readLine()) != null) {
- String[] item = data.split(",");
- if (item[0].equals("2")) {
- String sql = String.format("SET STORAGE GROUP TO %s", item[1]);
- statement.addBatch(sql);
- } else if (item[0].equals("0")) {
- String sql = String
- .format("CREATE TIMESERIES %s WITH DATATYPE=%s, ENCODING=%s", item[1], item[2],
- item[3]);
- statement.addBatch(sql);
- }
- count++;
- if (count > 10000) {
- statement.executeBatch();
- statement.clearBatch();
- count = 0;
- }
- }
- } catch (FileNotFoundException e) {
- logger.error("IoTDB post back receiver: cannot read the file {}.",
- schemaFromSenderPath.get(), e);
- } catch (IOException e) {
- logger.error("IoTDB post back receiver: cannot insert schema to IoTDB.", e);
- }
-
- statement.executeBatch();
- statement.clearBatch();
- } catch (SQLException | ClassNotFoundException e) {
- logger.error("IoTDB post back receiver: jdbc can not connect to IoTDB.", e);
- } finally {
- try {
- if (statement != null) {
- statement.close();
- }
- } catch (SQLException e) {
- logger.error("IoTDB post back receiver : can not close JDBC connection.", e);
- }
- }
+ /** sync metadata, include storage group and timeseries **/
+ syncMetadata();
} else {
File file = new File(schemaFromSenderPath.get());
if (!file.getParentFile().exists()) {
@@ -270,18 +239,84 @@ public class ServerServiceImpl implements ServerService.Iface {
logger.error("IoTDB post back receiver: cannot make schema file.", e);
}
}
- try {
- fos = new FileOutputStream(file, true);
- channel = fos.getChannel();
+ try (FileOutputStream fos = new FileOutputStream(file, true);
+ FileChannel channel = fos.getChannel()) {
channel.write(schema);
- channel.close();
- fos.close();
} catch (Exception e) {
- logger.error("IoTDB post back receiver: cannot write data to file.",
- e);
+ logger.error("IoTDB post back receiver: cannot write data to file.", e);
+ }
+ }
+
+ }
+
+ /**
+ * Sync metadata with sender
+ */
+ private void syncMetadata() {
+ if (new File(schemaFromSenderPath.get()).exists()) {
+ try (BufferedReader br = new BufferedReader(
+ new java.io.FileReader(schemaFromSenderPath.get()))) {
+ String metadataOperation;
+ while ((metadataOperation = br.readLine()) != null) {
+ operation(metadataOperation);
+ }
+ } catch (FileNotFoundException e) {
+ logger.error("IoTDB post back receiver: cannot read the file {}.",
+ schemaFromSenderPath.get(), e);
+ } catch (IOException e) {
+ logger.error("IoTDB post back receiver: cannot insert schema to IoTDB.", e);
+ } catch (Exception e) {
+ logger.error("IoTDB post back receiver: parse metadata operation failed.", e);
}
}
+ }
+ /**
+ * Operate metadata operation in MManager
+ *
+ * @param cmd metadata operation
+ */
+ private void operation(String cmd)
+ throws PathErrorException, IOException, MetadataArgsErrorException {
+ String[] args = cmd.trim().split(",");
+ switch (args[0]) {
+ case MetadataOperationType.ADD_PATH_TO_MTREE:
+ Map<String, String> props = null;
+ String[] kv;
+ props = new HashMap<>(args.length - 5 + 1, 1);
+ for (int k = 5; k < args.length; k++) {
+ kv = args[k].split("=");
+ props.put(kv[0], kv[1]);
+ }
+ metadataManger.addPathToMTree(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.deletePathFromMTree(args[1]);
+ break;
+ case MetadataOperationType.SET_STORAGE_LEVEL_TO_MTREE:
+ metadataManger.setStorageLevelToMTree(args[1]);
+ break;
+ case MetadataOperationType.ADD_A_PTREE:
+ metadataManger.addAPTree(args[1]);
+ break;
+ case MetadataOperationType.ADD_A_PATH_TO_PTREE:
+ metadataManger.addPathToPTree(args[1]);
+ break;
+ case MetadataOperationType.DELETE_PATH_FROM_PTREE:
+ metadataManger.deletePathFromPTree(args[1]);
+ break;
+ case MetadataOperationType.LINK_MNODE_TO_PTREE:
+ metadataManger.linkMNodeToPTree(args[1], args[2]);
+ break;
+ case MetadataOperationType.UNLINK_MNODE_FROM_PTREE:
+ metadataManger.unlinkMNodeFromPTree(args[1], args[2]);
+ break;
+ default:
+ logger.error("Unrecognizable command {}", cmd);
+ }
}
@Override
@@ -289,7 +324,7 @@ public class ServerServiceImpl implements ServerService.Iface {
getFileNodeInfo();
mergeData();
try {
- PostbackUtils.deleteFile(new File(postbackPath + this.uuid.get()));
+ SyncUtils.deleteFile(new File(postbackPath + this.uuid.get()));
} catch (IOException e) {
throw new TException(e);
}
@@ -297,10 +332,10 @@ public class ServerServiceImpl implements ServerService.Iface {
String backupPath = bufferWritePath + POSTBACK + File.separator;
if (new File(backupPath + this.uuid.get()).exists()
&& new File(backupPath + this.uuid.get()).list().length != 0) {
- // if does not exist, it means that the last time postback process failed, clear
+ // if does not exist, it means that the last time sync process failed, clear
// uuid data and receive the data again
try {
- PostbackUtils.deleteFile(new File(backupPath + this.uuid.get()));
+ SyncUtils.deleteFile(new File(backupPath + this.uuid.get()));
} catch (IOException e) {
throw new TException(e);
}
@@ -366,11 +401,13 @@ public class ServerServiceImpl implements ServerService.Iface {
fileNodeEndTime.get().put(fileTF.getAbsolutePath(), endTimeMap);
filesPath.add(fileTF.getAbsolutePath());
num++;
- logger.info(String
- .format("IoTDB receiver : Getting FileNode Info has complete : %d/%d", num,
- fileNum.get()));
+ if (logger.isInfoEnabled()) {
+ logger.info(String
+ .format("IoTDB receiver : Getting FileNode Info has complete : %d/%d", num,
+ fileNum.get()));
+ }
+ fileNodeMap.get().put(storageGroupPB.getName(), filesPath);
}
- fileNodeMap.get().put(storageGroupPB.getName(), filesPath);
}
}
@@ -379,78 +416,67 @@ public class ServerServiceImpl implements ServerService.Iface {
*/
@Override
public void mergeOldData(String filePath) throws TException {
- Set<String> timeseries = new HashSet<>();
- Statement statement = null;
+ Set<String> timeseriesSet = new HashSet<>();
TsFileSequenceReader reader = null;
- try (Connection connection = DriverManager.getConnection(
- String.format("jdbc:iotdb://localhost:%d/", config.getRpcPort()), "root",
- "root")) {
- Class.forName(JDBC_DRIVER_NAME);
- statement = connection.createStatement();
- int count = 0;
-
+ OverflowQPExecutor insertExecutor = new OverflowQPExecutor();
+ try {
+ /** use tsfile reader to get data **/
reader = new TsFileSequenceReader(filePath);
Map<String, TsDeviceMetadataIndex> deviceIdMap = reader.readFileMetadata().getDeviceMap();
- Iterator<String> it = deviceIdMap.keySet().iterator();
- while (it.hasNext()) {
- String deviceId = it.next(); // deviceId represent devices
- TsDeviceMetadataIndex deviceMetadataIndex = deviceIdMap.get(deviceId);
- TsDeviceMetadata deviceMetadata = reader.readTsDeviceMetaData(deviceMetadataIndex);
+ Iterator<Entry<String, TsDeviceMetadataIndex>> entryIterator = deviceIdMap.entrySet()
+ .iterator();
+ while (entryIterator.hasNext()) {
+ Entry<String, TsDeviceMetadataIndex> deviceMIEntry = entryIterator.next();
+ String deviceId = deviceMIEntry.getKey();
+ TsDeviceMetadataIndex deviceMI = deviceMIEntry.getValue();
+ TsDeviceMetadata deviceMetadata = reader.readTsDeviceMetaData(deviceMI);
List<ChunkGroupMetaData> rowGroupMetadataList = deviceMetadata.getChunkGroupMetaDataList();
- timeseries.clear();
- // firstly, get all timeseries in the same device
+ timeseriesSet.clear();
+ /** firstly, get all timeseries in the same device **/
for (ChunkGroupMetaData chunkGroupMetaData : rowGroupMetadataList) {
List<ChunkMetaData> chunkMetaDataList = chunkGroupMetaData
.getChunkMetaDataList();
for (ChunkMetaData chunkMetaData : chunkMetaDataList) {
String measurementUID = chunkMetaData.getMeasurementUid();
measurementUID = deviceId + "." + measurementUID;
- timeseries.add(measurementUID);
+ timeseriesSet.add(measurementUID);
}
}
- // secondly, use tsFile Reader to form SQL
-
+ /** Secondly, use tsFile Reader to form SQL **/
ReadOnlyTsFile readTsFile = new ReadOnlyTsFile(reader);
List<Path> paths = new ArrayList<>();
paths.clear();
- for (String timesery : timeseries) {
- paths.add(new Path(timesery));
+ for (String timeseries : timeseriesSet) {
+ paths.add(new Path(timeseries));
}
QueryExpression queryExpression = QueryExpression.create(paths, null);
QueryDataSet queryDataSet = readTsFile.query(queryExpression);
+ InsertPlan insertPlan;
while (queryDataSet.hasNext()) {
RowRecord record = queryDataSet.next();
List<Field> fields = record.getFields();
- String sqlFront = String.format("insert into %s(timestamp", deviceId);
- String sqlRear = String.format(") values(%d", record.getTimestamp());
+ List<String> measurementList = new ArrayList<>();
+ List<String> insertValues = new ArrayList<>();
for (int i = 0; i < fields.size(); i++) {
Field field = fields.get(i);
- if (field.toString() != "null") {
- sqlFront = String.format("%s,%s", sqlFront, paths.get(i).getMeasurement());
+ if (!field.isNull()) {
+ measurementList.add(paths.get(i).getMeasurement());
if (fields.get(i).getDataType() == TSDataType.TEXT) {
- sqlRear = String.format("%s,'%s'", sqlRear, field.toString());
+ insertValues.add(String.format("'%s'", field.toString()));
} else {
- sqlRear = String.format("%s,%s", sqlRear, field.toString());
+ insertValues.add(String.format("%s", field.toString()));
}
}
}
- String sql = String.format("%s%s)", sqlFront, sqlRear);
-
- statement.addBatch(sql);
- count++;
- if (count > 10000) {
- statement.executeBatch();
- statement.clearBatch();
- count = 0;
- }
+ insertPlan = new InsertPlan(deviceId, record.getTimestamp(), measurementList,
+ insertValues);
+ insertExecutor.processNonQuery(insertPlan);
}
}
- statement.executeBatch();
- statement.clearBatch();
} catch (IOException e) {
logger.error("IoTDB receiver can not parse tsfile into SQL", e);
- } catch (SQLException | ClassNotFoundException e) {
- logger.error("IoTDB post back receiver: jdbc cannot connect to IoTDB", e);
+ } catch (ProcessorException e) {
+ logger.error("Meet error while processing non-query.", e);
} finally {
try {
if (reader != null) {
@@ -459,84 +485,68 @@ public class ServerServiceImpl implements ServerService.Iface {
} catch (IOException e) {
logger.error("IoTDB receiver : Cannot close file stream {}", filePath, e);
}
- try {
- if (statement != null) {
- statement.close();
- }
- } catch (SQLException e) {
- logger.error("IoTDB receiver : Can not close JDBC connection", e);
- }
}
}
/**
* Insert those valid data in the tsfile into IoTDB
*
- * @param overlapFiles:files which are conflict with the postback file
+ * @param overlapFiles:files which are conflict with the sync file
*/
public void mergeOldData(String filePath, List<String> overlapFiles) {
- Set<String> timeseries = new HashSet<>();
+ Set<String> timeseriesList = new HashSet<>();
TsFileSequenceReader reader = null;
- Statement statement = null;
- try (Connection connection = DriverManager.getConnection(
- String.format("jdbc:iotdb://localhost:%d/", config.getRpcPort()), "root",
- "root")) {
- Class.forName(JDBC_DRIVER_NAME);
- statement = connection.createStatement();
- int count = 0;
-
+ OverflowQPExecutor insertExecutor = new OverflowQPExecutor();
+ try {
reader = new TsFileSequenceReader(filePath);
Map<String, TsDeviceMetadataIndex> deviceIdMap = reader.readFileMetadata().getDeviceMap();
Iterator<String> it = deviceIdMap.keySet().iterator();
while (it.hasNext()) {
- String key = it.next(); // key represent devices
- TsDeviceMetadataIndex deviceMetadataIndex = deviceIdMap.get(key);
- TsDeviceMetadata deviceMetadata = reader.readTsDeviceMetaData(deviceMetadataIndex);
+ String deviceID = it.next();
+ TsDeviceMetadataIndex deviceMI = deviceIdMap.get(deviceID);
+ TsDeviceMetadata deviceMetadata = reader.readTsDeviceMetaData(deviceMI);
List<ChunkGroupMetaData> chunkGroupMetaDataList = deviceMetadata
.getChunkGroupMetaDataList();
- timeseries.clear();
- // firstly, get all timeseries in the same device
+ timeseriesList.clear();
+ /** firstly, get all timeseries in the same device **/
for (ChunkGroupMetaData chunkGroupMetaData : chunkGroupMetaDataList) {
List<ChunkMetaData> chunkMetaDataList = chunkGroupMetaData.getChunkMetaDataList();
for (ChunkMetaData timeSeriesChunkMetaData : chunkMetaDataList) {
String measurementUID = timeSeriesChunkMetaData.getMeasurementUid();
- measurementUID = key + "." + measurementUID;
- timeseries.add(measurementUID);
+ measurementUID = deviceID + "." + measurementUID;
+ timeseriesList.add(measurementUID);
}
}
- // secondly, use tsFile Reader to form SQL
-
+ /** secondly, use tsFile Reader to form SQL **/
ReadOnlyTsFile readOnlyTsFile = new ReadOnlyTsFile(reader);
ArrayList<Path> paths = new ArrayList<>();
- for (String timesery : timeseries) { // compare data with one timesery in a round to get valid data
+ /** compare data with one timeseries in a round to get valid data **/
+ for (String timeseries : timeseriesList) {
paths.clear();
- paths.add(new Path(timesery));
- Map<String, String> originDataPoint = new HashMap<>();
- Map<String, String> newDataPoint = new HashMap<>();
- String sqlFormat = "insert into %s(timestamp,%s) values(%s,%s)";
+ paths.add(new Path(timeseries));
+ Map<InsertPlan, String> originDataPoint = new HashMap<>();
+ Map<InsertPlan, String> newDataPoint = new HashMap<>();
QueryExpression queryExpression = QueryExpression.create(paths, null);
QueryDataSet queryDataSet = readOnlyTsFile.query(queryExpression);
while (queryDataSet.hasNext()) {
RowRecord record = queryDataSet.next();
List<Field> fields = record.getFields();
- String sql;
- for (int i = 0; i < fields.size();
- i++) { // get all data with the timesery in the postback file
+ /** get all data with the timeseries in the sync file **/
+ for (int i = 0; i < fields.size(); i++) {
Field field = fields.get(i);
+ List<String> measurementList = new ArrayList<>();
if (!field.isNull()) {
- sql = String
- .format(sqlFormat, key, paths.get(i).getMeasurement(), record.getTimestamp(),
- "%s");
- if (field.getDataType() == TSDataType.TEXT) {
- newDataPoint.put(sql, "'" + field.toString() + "'");
- } else {
- newDataPoint.put(sql, field.toString());
- }
+ measurementList.add(paths.get(i).getMeasurement());
+ InsertPlan insertPlan = new InsertPlan(deviceID, record.getTimestamp(),
+ measurementList, new ArrayList<>());
+ newDataPoint.put(insertPlan,
+ field.getDataType() == TSDataType.TEXT ? String.format("'%s'", field.toString())
+ : field.toString());
}
}
}
- for (String overlapFile : overlapFiles) // get all data with the timesery in all overlap files.
- {
+ /** get all data with the timeseries in all overlap files. **/
+ for (String overlapFile : overlapFiles) {
TsFileSequenceReader inputOverlap = null;
try {
inputOverlap = new TsFileSequenceReader(overlapFile);
@@ -545,17 +555,17 @@ public class ServerServiceImpl implements ServerService.Iface {
while (queryDataSetOverlap.hasNext()) {
RowRecord recordOverlap = queryDataSetOverlap.next();
List<Field> fields = recordOverlap.getFields();
- String sql;
for (int i = 0; i < fields.size(); i++) {
Field field = fields.get(i);
+ List<String> measurementList = new ArrayList<>();
if (!field.isNull()) {
- sql = String.format(sqlFormat, key,
- paths.get(i).getMeasurement(), recordOverlap.getTimestamp(), "%s");
- if (field.getDataType() == TSDataType.TEXT) {
- originDataPoint.put(sql, "'" + field.toString() + "'");
- } else {
- originDataPoint.put(sql, field.toString());
- }
+ measurementList.add(paths.get(i).getMeasurement());
+ InsertPlan insertPlan = new InsertPlan(deviceID, recordOverlap.getTimestamp(),
+ measurementList, new ArrayList<>());
+ originDataPoint.put(insertPlan,
+ field.getDataType() == TSDataType.TEXT ? String
+ .format("'%s'", field.toString())
+ : field.toString());
}
}
}
@@ -565,57 +575,42 @@ public class ServerServiceImpl implements ServerService.Iface {
}
}
}
- if (originDataPoint
- .isEmpty()) { // If there has no overlap data with the timesery, inserting all data in the postback file
- for (Map.Entry<String, String> entry : newDataPoint.entrySet()) {
- String sql = String.format(entry.getKey(), entry.getValue());
- statement.addBatch(sql);
- count++;
- if (count > 10000) {
- statement.executeBatch();
- statement.clearBatch();
- count = 0;
- }
+
+ /** If there has no overlap data with the timeseries, inserting all data in the sync file **/
+ if (originDataPoint.isEmpty()) {
+ for (Map.Entry<InsertPlan, String> entry : newDataPoint.entrySet()) {
+ InsertPlan insertPlan = entry.getKey();
+ List<String> insertValues = new ArrayList<>();
+ insertValues.add(entry.getValue());
+ insertPlan.setValues(insertValues);
+ insertExecutor.processNonQuery(insertPlan);
}
- } else { // Compare every data to get valid data
- for (Map.Entry<String, String> entry : newDataPoint.entrySet()) {
+ } else {
+ /** Compare every data to get valid data **/
+ for (Map.Entry<InsertPlan, String> entry : newDataPoint.entrySet()) {
if (!originDataPoint.containsKey(entry.getKey())
|| (originDataPoint.containsKey(entry.getKey())
&& !originDataPoint.get(entry.getKey()).equals(entry.getValue()))) {
- String sql = String.format(entry.getKey(), entry.getValue());
- statement.addBatch(sql);
- count++;
- if (count > 10000) {
- statement.executeBatch();
- statement.clearBatch();
- count = 0;
- }
+ InsertPlan insertPlan = entry.getKey();
+ List<String> insertValues = new ArrayList<>();
+ insertValues.add(entry.getValue());
+ insertPlan.setValues(insertValues);
+ insertExecutor.processNonQuery(insertPlan);
}
}
}
}
}
- statement.executeBatch();
- statement.clearBatch();
- } catch (SQLException e) {
- logger.error("IoTDB post back receiver: sql cannot execute successfully in IoTDB", e);
- } catch (ClassNotFoundException e) {
- logger.error("IoTDB post back receiver: jdbc cannot connect to IoTDB", e);
} catch (IOException e) {
logger.error("IoTDB receiver can not parse tsfile into SQL", e);
+ } catch (ProcessorException e) {
+ logger.error("Meet error while processing non-query.", e);
} finally {
try {
reader.close();
} catch (IOException e) {
logger.error("IoTDB receiver : Cannot close file stream {}", filePath, e);
}
- try {
- if (statement != null) {
- statement.close();
- }
- } catch (SQLException e) {
- logger.error("IoTDB receiver : Can not close JDBC connection", e);
- }
}
}
@@ -685,8 +680,10 @@ public class ServerServiceImpl implements ServerService.Iface {
}
num++;
- logger.info(String
- .format("IoTDB receiver : Merging files has completed : %d/%d", num, fileNum.get()));
+ if (logger.isInfoEnabled()) {
+ logger.info(String
+ .format("IoTDB receiver : Merging files has completed : %d/%d", num, fileNum.get()));
+ }
}
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileManager.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
similarity index 67%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileManager.java
rename to iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
index d46a308..f0a3b66 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileManager.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.sender;
+package org.apache.iotdb.db.sync.sender;
import java.io.BufferedReader;
import java.io.BufferedWriter;
@@ -29,27 +29,43 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.postback.conf.PostBackSenderConfig;
-import org.apache.iotdb.db.postback.conf.PostBackSenderDescriptor;
+import org.apache.iotdb.db.sync.conf.Constans;
+import org.apache.iotdb.db.sync.conf.SyncSenderConfig;
+import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * The class is to pick up which files need to postback.
+ * FileManager is used to pick up those tsfiles need to sync.
*
- * @author lta
+ * @author Tianan Li
*/
public class FileManager {
private static final Logger LOGGER = LoggerFactory.getLogger(FileManager.class);
- private Map<String, Set<String>> sendingFiles = new HashMap<>();
+
+ /**
+ * Files that need to be synchronized
+ **/
+ private Map<String, Set<String>> validAllFiles = new HashMap<>();
+
+ /**
+ * All tsfiles in last synchronization process
+ **/
private Set<String> lastLocalFiles = new HashSet<>();
+
+ /**
+ * All tsfiles in data directory
+ **/
private Map<String, Set<String>> currentLocalFiles = new HashMap<>();
- private PostBackSenderConfig postbackConfig = PostBackSenderDescriptor.getInstance().getConfig();
- private IoTDBConfig tsfileConfig = IoTDBDescriptor.getInstance().getConfig();
+
+ private SyncSenderConfig syncConfig = SyncSenderDescriptor.getInstance().getConfig();
+
+ private IoTDBConfig systemConfig = IoTDBDescriptor.getInstance().getConfig();
+
+ private static final String RESTORE_SUFFIX = ".restore";
private FileManager() {
}
@@ -59,30 +75,30 @@ public class FileManager {
}
/**
- * initialize FileManager.
+ * Initialize FileManager.
*/
- public void init() {
- sendingFiles.clear();
+ public void init() throws IOException {
+ validAllFiles.clear();
lastLocalFiles.clear();
currentLocalFiles.clear();
- getLastLocalFileList(postbackConfig.getLastFileInfo());
- getCurrentLocalFileList(tsfileConfig.getBufferWriteDirs());
- getSendingFileList();
+ getLastLocalFileList(syncConfig.getLastFileInfo());
+ getCurrentLocalFileList(systemConfig.getBufferWriteDirs());
+ getValidFileList();
}
/**
- * get sending file list.
+ * get files that needs to be synchronized
*/
- public void getSendingFileList() {
+ public void getValidFileList() {
for (Entry<String, Set<String>> entry : currentLocalFiles.entrySet()) {
for (String path : entry.getValue()) {
if (!lastLocalFiles.contains(path)) {
- sendingFiles.get(entry.getKey()).add(path);
+ validAllFiles.get(entry.getKey()).add(path);
}
}
}
- LOGGER.info("IoTDB sender : Sender has got list of sending files.");
- for (Entry<String, Set<String>> entry : sendingFiles.entrySet()) {
+ LOGGER.info("acquire list of valid files.");
+ for (Entry<String, Set<String>> entry : validAllFiles.entrySet()) {
for (String path : entry.getValue()) {
LOGGER.info(path);
currentLocalFiles.get(entry.getKey()).remove(path);
@@ -95,16 +111,16 @@ public class FileManager {
*
* @param path path
*/
- public void getLastLocalFileList(String path) {
+ public void getLastLocalFileList(String path) throws IOException {
Set<String> fileList = new HashSet<>();
File file = new File(path);
if (!file.exists()) {
try {
if (!file.createNewFile()) {
- LOGGER.error("IoTDB post back sender: cannot create file {}", file.getAbsoluteFile());
+ LOGGER.error("cannot create file {}", file.getAbsoluteFile());
}
} catch (IOException e) {
- LOGGER.error("IoTDB post back sender: cannot get last local file list", e);
+ throw new IOException("cannot get last local file list", e);
}
} else {
try (BufferedReader bf = new BufferedReader(new FileReader(file))) {
@@ -113,12 +129,11 @@ public class FileManager {
fileList.add(fileName);
}
} catch (IOException e) {
- LOGGER.error(
- "IoTDB post back sender: cannot get last local file list when reading file {}.",
- postbackConfig.getLastFileInfo(), e);
+ LOGGER.error("cannot get last local file list when reading file {}.",
+ syncConfig.getLastFileInfo());
+ throw new IOException(e);
}
}
-
lastLocalFiles = fileList;
}
@@ -134,17 +149,17 @@ public class FileManager {
}
File[] listFiles = new File(path).listFiles();
for (File storageGroup : listFiles) {
- if (storageGroup.isDirectory() && !storageGroup.getName().equals("postback")) {
+ if (storageGroup.isDirectory() && !storageGroup.getName().equals(Constans.SYNC)) {
if (!currentLocalFiles.containsKey(storageGroup.getName())) {
- currentLocalFiles.put(storageGroup.getName(), new HashSet<String>());
+ currentLocalFiles.put(storageGroup.getName(), new HashSet<>());
}
- if (!sendingFiles.containsKey(storageGroup.getName())) {
- sendingFiles.put(storageGroup.getName(), new HashSet<String>());
+ if (!validAllFiles.containsKey(storageGroup.getName())) {
+ validAllFiles.put(storageGroup.getName(), new HashSet<>());
}
File[] files = storageGroup.listFiles();
for (File file : files) {
- if (!file.getAbsolutePath().endsWith(".restore") && !new File(
- file.getAbsolutePath() + ".restore").exists()) {
+ if (!file.getPath().endsWith(RESTORE_SUFFIX) && !new File(
+ file.getPath() + RESTORE_SUFFIX).exists()) {
currentLocalFiles.get(storageGroup.getName()).add(file.getAbsolutePath());
}
}
@@ -166,12 +181,12 @@ public class FileManager {
}
}
} catch (IOException e) {
- LOGGER.error("IoTDB post back sender: cannot back up now local file info", e);
+ LOGGER.error("cannot back up now local file info", e);
}
}
- public Map<String, Set<String>> getSendingFiles() {
- return sendingFiles;
+ public Map<String, Set<String>> getValidAllFiles() {
+ return validAllFiles;
}
public Set<String> getLastLocalFiles() {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileSender.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSender.java
similarity index 58%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileSender.java
rename to iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSender.java
index 0ca03c0..8e8ada9 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/sender/FileSender.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSender.java
@@ -16,51 +16,52 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.sender;
+package org.apache.iotdb.db.sync.sender;
+import java.io.IOException;
import java.util.Set;
+import org.apache.iotdb.db.exception.SyncConnectionException;
/**
- * FileSender defines the methods of a sender in postback module.
- * @author lta
+ * FileSender defines the methods of a sender in sync module.
+ *
+ * @author Tianan Li
*/
public interface FileSender {
/**
* Connect to server.
*/
- void connectToReceiver(String serverIp, int serverPort);
+ void establishConnection(String serverIp, int serverPort) throws SyncConnectionException;
/**
* Transfer UUID to receiver.
*/
- boolean transferUUID(String uuidPath);
+ boolean confirmIdentity(String uuidPath) throws SyncConnectionException, IOException;
/**
* Make file snapshots before sending files.
*/
- Set<String> makeFileSnapshot(Set<String> sendingFileList);
+ Set<String> makeFileSnapshot(Set<String> validFiles) throws IOException;
/**
* Send schema file to receiver.
*/
- void sendSchema(String schemaPath);
+ void syncSchema() throws SyncConnectionException;
/**
- * For each file in fileList, send it to receiver side.
- *
- * @param fileSnapshotList snapshot file list to send
+ * For all valid files, send it to receiver side and load these data in receiver.
*/
- void transferData(Set<String> fileSnapshotList);
+ void syncAllData() throws SyncConnectionException;
/**
* Close the socket after sending files.
*/
- boolean afterSending();
+ boolean afterSynchronization() throws SyncConnectionException;
/**
- * Execute a postback task.
+ * Execute a sync task.
*/
- void postback();
+ void sync() throws SyncConnectionException, IOException;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
new file mode 100644
index 0000000..ea34a29
--- /dev/null
+++ b/iotdb/src/main/java/org/apache/iotdb/db/sync/sender/FileSenderImpl.java
@@ -0,0 +1,516 @@
+/**
+ * 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.sync.sender;
+
+import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.iotdb.db.concurrent.ThreadName;
+import org.apache.iotdb.db.exception.SyncConnectionException;
+import org.apache.iotdb.db.sync.conf.Constans;
+import org.apache.iotdb.db.sync.conf.SyncSenderConfig;
+import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
+import org.apache.iotdb.db.sync.receiver.ServerService;
+import org.apache.iotdb.db.utils.SyncUtils;
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * FileSenderImpl is used to transfer tsfiles that needs to sync to receiver.
+ *
+ * @author Tianan Li
+ */
+public class FileSenderImpl implements FileSender {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FileSenderImpl.class);
+ private TTransport transport;
+ private ServerService.Client serviceClient;
+ private List<String> schema = new ArrayList<>();
+
+ /**
+ * Mark the identity of sender
+ **/
+ private String uuid;
+
+ /**
+ * Monitor sync status
+ **/
+ private Thread syncMonitor;
+
+ /**
+ * Files that need to be synchronized
+ */
+ private Map<String, Set<String>> validAllFiles;
+
+ /**
+ * All tsfiles in data directory
+ **/
+ private Map<String, Set<String>> currentLocalFiles;
+
+ /**
+ * Mark the start time of last sync
+ **/
+ private Date lastSyncTime = new Date();
+
+ /**
+ * If true, sync is in execution.
+ **/
+ private volatile boolean syncStatus = false;
+
+ /**
+ * Key means storage group, Set means corresponding tsfiles
+ **/
+ private Map<String, Set<String>> validFileSnapshot = new HashMap<>();
+
+ private FileManager fileManager = FileManager.getInstance();
+ private SyncSenderConfig config = SyncSenderDescriptor.getInstance().getConfig();
+
+ /**
+ * Monitor sync status.
+ */
+ private final Runnable monitorSyncStatus = () -> {
+ Date oldTime = new Date();
+ while (true) {
+ if (Thread.interrupted()) {
+ break;
+ }
+ Date currentTime = new Date();
+ if (currentTime.getTime() / 1000 == oldTime.getTime() / 1000) {
+ continue;
+ }
+ if ((currentTime.getTime() - lastSyncTime.getTime())
+ % (config.getUploadCycleInSeconds() * 1000) == 0) {
+ oldTime = currentTime;
+ if (syncStatus) {
+ LOGGER.info("Sync process is in execution!");
+ }
+ }
+ }
+ };
+
+ private FileSenderImpl() {
+ }
+
+ public static final FileSenderImpl getInstance() {
+ return InstanceHolder.INSTANCE;
+ }
+
+ /**
+ * Create a sender and sync files to the receiver.
+ *
+ * @param args not used
+ */
+ public static void main(String[] args)
+ throws InterruptedException, IOException, SyncConnectionException {
+ Thread.currentThread().setName(ThreadName.SYNC_CLIENT.getName());
+ FileSenderImpl fileSenderImpl = new FileSenderImpl();
+ fileSenderImpl.verifyPort();
+ fileSenderImpl.startMonitor();
+ fileSenderImpl.timedTask();
+ }
+
+ /**
+ * Start Monitor Thread
+ */
+ public void startMonitor() {
+ syncMonitor = new Thread(monitorSyncStatus, ThreadName.SYNC_MONITOR.getName());
+ syncMonitor.setDaemon(true);
+ syncMonitor.start();
+ }
+
+ /**
+ * Start sync task in a certain time.
+ */
+ public void timedTask() throws InterruptedException, SyncConnectionException, IOException {
+ sync();
+ lastSyncTime = new Date();
+ Date currentTime;
+ while (true) {
+ if (Thread.interrupted()) {
+ break;
+ }
+ Thread.sleep(2000);
+ currentTime = new Date();
+ if (currentTime.getTime() - lastSyncTime.getTime()
+ > config.getUploadCycleInSeconds() * 1000) {
+ lastSyncTime = currentTime;
+ sync();
+ }
+ }
+ }
+
+ /**
+ * Execute a sync task.
+ */
+ @Override
+ public void sync() throws SyncConnectionException, IOException {
+
+ //1. Clear old snapshots if necessary
+ for (String snapshotPath : config.getSnapshotPaths()) {
+ if (new File(snapshotPath).exists() && new File(snapshotPath).list().length != 0) {
+ /** It means that the last task of sync does not succeed! Clear the files and start to sync again **/
+ try {
+ SyncUtils.deleteFile(new File(snapshotPath));
+ } catch (IOException e) {
+ LOGGER.error("can not delete file {}", snapshotPath);
+ throw new IOException(e);
+ }
+ }
+ }
+
+ syncStatus = true;
+
+ // 2. Connect to sync server and Confirm Identity
+ establishConnection(config.getServerIp(), config.getServerPort());
+ if (!confirmIdentity(config.getUuidPath())) {
+ LOGGER.error("Sorry, you do not have the permission to connect to sync receiver.");
+ syncStatus = false;
+ return;
+ }
+
+ // 3. Acquire valid files and check
+ fileManager.init();
+ validAllFiles = fileManager.getValidAllFiles();
+ currentLocalFiles = fileManager.getCurrentLocalFiles();
+ if (SyncUtils.isEmpty(validAllFiles)) {
+ LOGGER.info("There has no file to sync !");
+ syncStatus = false;
+ return;
+ }
+
+ // 4. Create snapshot
+ for (Entry<String, Set<String>> entry : validAllFiles.entrySet()) {
+ validFileSnapshot.put(entry.getKey(), makeFileSnapshot(entry.getValue()));
+ }
+
+ // 5. Sync schema
+ syncSchema();
+
+ // 6. Sync data
+ syncAllData();
+
+ // 7. clear snapshot
+ for (String snapshotPath : config.getSnapshotPaths()) {
+ try {
+ SyncUtils.deleteFile(new File(snapshotPath));
+ } catch (IOException e) {
+ LOGGER.error("can not delete snapshot", e);
+ }
+ }
+
+ // 8. notify receiver that synchronization finish
+ // At this point the synchronization has finished even if connection fails
+ try {
+ serviceClient.afterReceiving();
+ } catch (TException e) {
+ LOGGER.error("unable to connect to receiver ", e);
+ }
+ transport.close();
+ LOGGER.info("sync process has finished");
+ syncStatus = false;
+ }
+
+ @Override
+ public void syncAllData() throws SyncConnectionException {
+ for (Entry<String, Set<String>> entry : validAllFiles.entrySet()) {
+ Set<String> validFiles = entry.getValue();
+ Set<String> validSnapshot = validFileSnapshot.get(entry.getKey());
+ if (validSnapshot.isEmpty()) {
+ continue;
+ }
+ LOGGER.info("sync process starts to transfer data of storage group {}", entry.getKey());
+ try {
+ serviceClient.init(entry.getKey());
+ } catch (TException e) {
+ throw new SyncConnectionException("unable to connect to receiver", e);
+ }
+ syncData(validSnapshot);
+ if (afterSynchronization()) {
+ currentLocalFiles.get(entry.getKey()).addAll(validFiles);
+ fileManager.setCurrentLocalFiles(currentLocalFiles);
+ fileManager.backupNowLocalFileInfo(config.getLastFileInfo());
+ LOGGER.info("sync process has finished storage group {}.", entry.getKey());
+ } else {
+ throw new SyncConnectionException(
+ "receiver cannot sync data, abandon this synchronization");
+ }
+ }
+ }
+
+ /**
+ * Establish a connection between the sender and the receiver.
+ *
+ * @param serverIp the ip address of the receiver
+ * @param serverPort must be same with port receiver set.
+ */
+ @Override
+ public void establishConnection(String serverIp, int serverPort) throws SyncConnectionException {
+ transport = new TSocket(serverIp, serverPort);
+ TProtocol protocol = new TBinaryProtocol(transport);
+ serviceClient = new ServerService.Client(protocol);
+ try {
+ transport.open();
+ } catch (TTransportException e) {
+ syncStatus = false;
+ LOGGER.error("cannot connect to server");
+ throw new SyncConnectionException(e);
+ }
+ }
+
+ /**
+ * UUID marks the identity of sender for receiver.
+ */
+ @Override
+ public boolean confirmIdentity(String uuidPath) throws SyncConnectionException, IOException {
+ File file = new File(uuidPath);
+ if (!file.getParentFile().exists()) {
+ file.getParentFile().mkdirs();
+ }
+ if (!file.exists()) {
+ try (FileOutputStream out = new FileOutputStream(file)) {
+ if (!file.createNewFile()) {
+ LOGGER.error("cannot create file {}", file.getPath());
+ }
+ uuid = generateUUID();
+ out.write(uuid.getBytes());
+ } catch (IOException e) {
+ LOGGER.error("cannot write UUID to file {}", file.getPath());
+ throw new IOException(e);
+ }
+ } else {
+ try (BufferedReader bf = new BufferedReader((new FileReader(uuidPath)))) {
+ uuid = bf.readLine();
+ } catch (IOException e) {
+ LOGGER.error("cannot read UUID from file{}", file.getPath());
+ throw new IOException(e);
+ }
+ }
+ boolean legalConnection;
+ try {
+ legalConnection = serviceClient.getUUID(uuid,
+ InetAddress.getLocalHost().getHostAddress());
+ } catch (Exception e) {
+ LOGGER.error("cannot confirm identity with receiver");
+ throw new SyncConnectionException(e);
+ }
+ return legalConnection;
+ }
+
+ private String generateUUID() {
+ return Constans.SYNC + UUID.randomUUID().toString().replaceAll("-", "");
+ }
+
+ /**
+ * Create snapshots for valid files.
+ */
+ @Override
+ public Set<String> makeFileSnapshot(Set<String> validFiles) throws IOException {
+ Set<String> validFilesSnapshot = new HashSet<>();
+ try {
+ for (String filePath : validFiles) {
+ String snapshotFilePath = SyncUtils.getSnapshotFilePath(filePath);
+ validFilesSnapshot.add(snapshotFilePath);
+ File newFile = new File(snapshotFilePath);
+ if (!newFile.getParentFile().exists()) {
+ newFile.getParentFile().mkdirs();
+ }
+ Path link = FileSystems.getDefault().getPath(snapshotFilePath);
+ Path target = FileSystems.getDefault().getPath(filePath);
+ Files.createLink(link, target);
+ }
+ } catch (IOException e) {
+ LOGGER.error("can not make fileSnapshot");
+ throw new IOException(e);
+ }
+ return validFilesSnapshot;
+ }
+
+ /**
+ * Transfer data of a storage group to receiver.
+ *
+ * @param fileSnapshotList list of sending snapshot files in a storage group.
+ */
+ public void syncData(Set<String> fileSnapshotList) throws SyncConnectionException {
+ try {
+ int successNum = 0;
+ for (String snapshotFilePath : fileSnapshotList) {
+ successNum++;
+ File file = new File(snapshotFilePath);
+ List<String> filePathSplit = new ArrayList<>();
+ String os = System.getProperty("os.name");
+ if (os.toLowerCase().startsWith("windows")) {
+ String[] name = snapshotFilePath.split(File.separator + File.separator);
+ filePathSplit.add(name[name.length - 2]);
+ filePathSplit.add(name[name.length - 1]);
+ } else {
+ String[] name = snapshotFilePath.split(File.separator);
+ filePathSplit.add(name[name.length - 2]);
+ filePathSplit.add(name[name.length - 1]);
+ }
+ while (true) {
+ // Sync all data to receiver
+ byte[] buffer = new byte[Constans.DATA_CHUNK_SIZE];
+ int data;
+ try (FileInputStream fis = new FileInputStream(file)) {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream(Constans.DATA_CHUNK_SIZE);
+ while ((data = fis.read(buffer)) != -1) { // cut the file into pieces to send
+ bos.write(buffer, 0, data);
+ ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
+ bos.reset();
+ serviceClient.startReceiving(null, filePathSplit, buffToSend, 1);
+ }
+ bos.close();
+ }
+
+ // Get md5 of the file.
+ MessageDigest md = MessageDigest.getInstance("MD5");
+ try (FileInputStream fis = new FileInputStream(file)) {
+ while ((data = fis.read(buffer)) != -1) {
+ md.update(buffer, 0, data);
+ }
+ }
+
+ // the file is sent successfully
+ String md5OfSender = (new BigInteger(1, md.digest())).toString(16);
+ String md5OfReceiver = serviceClient.startReceiving(md5OfSender, filePathSplit,
+ null, 0);
+ if (md5OfSender.equals(md5OfReceiver)) {
+ LOGGER.info("receiver has received {} successfully.", snapshotFilePath);
+ break;
+ }
+ }
+ if (LOGGER.isInfoEnabled()) {
+ LOGGER.info(String.format("Task of synchronization has completed %d/%d.", successNum,
+ fileSnapshotList.size()));
+ }
+ }
+ } catch (Exception e) {
+ throw new SyncConnectionException("cannot sync data with receiver.", e);
+ }
+ }
+
+ /**
+ * Sync schema with receiver.
+ */
+ @Override
+ public void syncSchema() throws SyncConnectionException {
+ try (FileInputStream fis = new FileInputStream(new File(config.getSchemaPath()))) {
+ int mBufferSize = 4 * 1024 * 1024;
+ ByteArrayOutputStream bos = new ByteArrayOutputStream(mBufferSize);
+ byte[] buffer = new byte[mBufferSize];
+ int n;
+ while ((n = fis.read(buffer)) != -1) { // cut the file into pieces to send
+ bos.write(buffer, 0, n);
+ ByteBuffer buffToSend = ByteBuffer.wrap(bos.toByteArray());
+ bos.reset();
+ // 1 represents there is still schema buffer to send.
+ serviceClient.getSchema(buffToSend, 1);
+ }
+ bos.close();
+ // 0 represents the schema file has been transferred completely.
+ serviceClient.getSchema(null, 0);
+ } catch (Exception e) {
+ LOGGER.error("cannot sync schema ", e);
+ throw new SyncConnectionException(e);
+ }
+ }
+
+ @Override
+ public boolean afterSynchronization() throws SyncConnectionException {
+ boolean successOrNot;
+ try {
+ successOrNot = serviceClient.merge();
+ } catch (TException e) {
+ throw new SyncConnectionException(
+ "can not finish sync process because sync receiver has broken down.", e);
+ }
+ return successOrNot;
+ }
+
+ /**
+ * The method is to verify whether the client port is bind or not, ensuring that only one client
+ * is running.
+ */
+ private void verifyPort() throws IOException {
+ try {
+ Socket socket = new Socket("localhost", config.getClientPort());
+ socket.close();
+ LOGGER.error("The sync client has been started!");
+ System.exit(0);
+ } catch (IOException e) {
+ try (ServerSocket listenerSocket = new ServerSocket(config.getClientPort())) {
+ Thread listener = new Thread(() -> {
+ while (true) {
+ try {
+ listenerSocket.accept();
+ } catch (IOException e2) {
+ LOGGER.error("IoTDB sync sender: unable to listen to port{}",
+ config.getClientPort(), e2);
+ }
+ }
+ });
+ listener.start();
+ } catch (IOException e1) {
+ LOGGER.error("unable to listen to port{}", config.getClientPort());
+ throw new IOException();
+ }
+ }
+ }
+
+ private static class InstanceHolder {
+
+ private static final FileSenderImpl INSTANCE = new FileSenderImpl();
+ }
+
+ public void setConfig(SyncSenderConfig config) {
+ this.config = config;
+ }
+
+ public List<String> getSchema() {
+ return schema;
+ }
+}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/utils/PostbackUtils.java b/iotdb/src/main/java/org/apache/iotdb/db/utils/SyncUtils.java
similarity index 70%
rename from iotdb/src/main/java/org/apache/iotdb/db/utils/PostbackUtils.java
rename to iotdb/src/main/java/org/apache/iotdb/db/utils/SyncUtils.java
index a23e0be..998aff2 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/utils/PostbackUtils.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/utils/SyncUtils.java
@@ -21,24 +21,25 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
-import org.apache.iotdb.db.postback.conf.PostBackSenderDescriptor;
+import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
/**
* @author lta
*/
-public class PostbackUtils {
+public class SyncUtils {
- private PostbackUtils(){}
+ private static final String IP_SEPARATOR = "\\.";
- private static String[] snapshotPaths = PostBackSenderDescriptor.getInstance()
+ private static String[] snapshotPaths = SyncSenderDescriptor.getInstance()
.getConfig().getSnapshotPaths();
+ private SyncUtils() {
+ }
+
/**
- * This method is to get a snapshot file seriesPath according to a tsfile seriesPath. Due to multiple directories,
- * it's necessary to make a snapshot in the same disk. It's used by postback sender.
- *
- * @param filePath
- * @return
+ * This method is to get a snapshot file seriesPath according to a tsfile seriesPath. Due to
+ * multiple directories, it's necessary to make a snapshot in the same disk. It's used by sync
+ * sender.
*/
public static String getSnapshotFilePath(String filePath) {
String[] name;
@@ -46,22 +47,23 @@ public class PostbackUtils {
String os = System.getProperty("os.name");
if (os.toLowerCase().startsWith("windows")) {
name = filePath.split(File.separator + File.separator);
- relativeFilePath =
- "data" + File.separator + name[name.length - 2] + File.separator + name[name.length - 1];
+ relativeFilePath = name[name.length - 2] + File.separator + name[name.length - 1];
} else {
name = filePath.split(File.separator);
- relativeFilePath =
- "data" + File.separator + name[name.length - 2] + File.separator + name[name.length - 1];
+ relativeFilePath = name[name.length - 2] + File.separator + name[name.length - 1];
}
String bufferWritePath = name[0];
for (int i = 1; i < name.length - 2; i++) {
- bufferWritePath = bufferWritePath + File.separator + name[i];
+ bufferWritePath = bufferWritePath + File.separatorChar + name[i];
}
for (String snapshotPath : snapshotPaths) {
if (snapshotPath.startsWith(bufferWritePath)) {
if (!new File(snapshotPath).exists()) {
new File(snapshotPath).mkdir();
}
+ if(snapshotPath.length() > 0 && snapshotPath.charAt(snapshotPath.length()-1)!=File.separatorChar){
+ snapshotPath = snapshotPath + File.separatorChar;
+ }
return snapshotPath + relativeFilePath;
}
}
@@ -69,10 +71,7 @@ public class PostbackUtils {
}
/**
- * Verify sending list is empty or not It's used by postback sender.
- *
- * @param sendingFileList
- * @return
+ * Verify sending list is empty or not It's used by sync sender.
*/
public static boolean isEmpty(Map<String, Set<String>> sendingFileList) {
for (Entry<String, Set<String>> entry : sendingFileList.entrySet()) {
@@ -84,11 +83,8 @@ public class PostbackUtils {
}
/**
- * Verify IP address with IP white list which contains more than one IP segment. It's used by postback sender.
- *
- * @param ipWhiteList
- * @param ipAddress
- * @return
+ * Verify IP address with IP white list which contains more than one IP segment. It's used by sync
+ * sender.
*/
public static boolean verifyIPSegment(String ipWhiteList, String ipAddress) {
String[] ipSegments = ipWhiteList.split(",");
@@ -104,43 +100,43 @@ public class PostbackUtils {
/**
* Verify IP address with IP segment.
- *
- * @param ipSegment
- * @param ipAddress
- * @param subnetMark
- * @return
*/
private static boolean verifyIP(String ipSegment, String ipAddress, int subnetMark) {
- String ipSegmentBinary = "";
- String ipAddressBinary = "";
- String[] ipSplits = ipSegment.split("\\.");
+ String ipSegmentBinary;
+ String ipAddressBinary;
+ String[] ipSplits = ipSegment.split(IP_SEPARATOR);
DecimalFormat df = new DecimalFormat("00000000");
StringBuilder ipSegmentBuilder = new StringBuilder();
for (String IPsplit : ipSplits) {
- ipSegmentBuilder.append(String.valueOf(df.format(
- Integer.parseInt(Integer.toBinaryString(Integer.parseInt(IPsplit))))));
+ ipSegmentBuilder.append(df.format(
+ Integer.parseInt(Integer.toBinaryString(Integer.parseInt(IPsplit)))));
}
ipSegmentBinary = ipSegmentBuilder.toString();
ipSegmentBinary = ipSegmentBinary.substring(0, subnetMark);
- ipSplits = ipAddress.split("\\.");
+ ipSplits = ipAddress.split(IP_SEPARATOR);
StringBuilder ipAddressBuilder = new StringBuilder();
for (String IPsplit : ipSplits) {
- ipAddressBuilder.append(String.valueOf(df.format(
- Integer.parseInt(Integer.toBinaryString(Integer.parseInt(IPsplit))))));
+ ipAddressBuilder.append(df.format(
+ Integer.parseInt(Integer.toBinaryString(Integer.parseInt(IPsplit)))));
}
ipAddressBinary = ipAddressBuilder.toString();
ipAddressBinary = ipAddressBinary.substring(0, subnetMark);
return ipAddressBinary.equals(ipSegmentBinary);
}
+ /**
+ * Remove all files under this folder recursively
+ *
+ * @param file folder file
+ */
public static void deleteFile(File file) throws IOException {
if (!file.exists()) {
return;
}
if (file.isFile() || Objects.requireNonNull(file.list()).length == 0) {
- if (!file.delete()){
+ if (!file.delete()) {
throw new IOException(
- String.format("Cannot delete file : %s", file.getPath()));
+ String.format("cannot delete file : %s", file.getPath()));
}
} else {
File[] files = file.listFiles();
@@ -149,7 +145,7 @@ public class PostbackUtils {
deleteFile(f);
if (!f.delete()) {
throw new IOException(
- String.format("Cannot delete file : %s", f.getPath()));
+ String.format("cannot delete file : %s", f.getPath()));
}
}
}
diff --git a/iotdb/src/main/thrift/ServerService.thrift b/iotdb/src/main/thrift/SyncServerService.thrift
similarity index 79%
rename from iotdb/src/main/thrift/ServerService.thrift
rename to iotdb/src/main/thrift/SyncServerService.thrift
index cdcc47c..d633d25 100755
--- a/iotdb/src/main/thrift/ServerService.thrift
+++ b/iotdb/src/main/thrift/SyncServerService.thrift
@@ -16,19 +16,23 @@
* specific language governing permissions and limitations
* under the License.
*/
- namespace java org.apache.iotdb.db.postback.receiver
+namespace java org.apache.iotdb.db.sync.receiver
typedef i32 int
typedef i16 short
typedef i64 long
+
+enum SYNC_STATUS {
+ SUCCESS_STATUS,
+ SYNC_STATUS
+
+}
+
service ServerService{
bool getUUID(1:string uuid, 2:string address)
- string startReceiving(1:string md5, 2:list<string> filename, 3:binary buff, 4:int status)
- void getFileNodeInfo()
- void mergeOldData(1:string path)
- void mergeData()
void getSchema(1:binary buff, 2:int status)
- bool merge()
+ string receiveData(1:string md5, 2:list<string> filename, 3:binary buff, 4:int status)
+ bool load()
void afterReceiving()
void init(1:string storageGroup)
}
\ No newline at end of file
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/filenodev2/FileNodeManagerBenchmark.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/filenodev2/FileNodeManagerBenchmark.java
index 5829f0e..55ee355 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/filenodev2/FileNodeManagerBenchmark.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/filenodev2/FileNodeManagerBenchmark.java
@@ -26,7 +26,7 @@ import org.apache.iotdb.db.exception.FileNodeManagerException;
import org.apache.iotdb.db.exception.MetadataArgsErrorException;
import org.apache.iotdb.db.exception.PathErrorException;
import org.apache.iotdb.db.metadata.MManager;
-import org.apache.iotdb.db.postback.utils.RandomNum;
+import org.apache.iotdb.db.sync.test.RandomNum;
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;
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/postback/sender/FileManagerTest.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
similarity index 95%
rename from iotdb/src/test/java/org/apache/iotdb/db/postback/sender/FileManagerTest.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
index f808a87..a2dd6c6 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/postback/sender/FileManagerTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/FileManagerTest.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.sender;
+package org.apache.iotdb.db.sync.sender;
import java.io.File;
import java.io.IOException;
@@ -26,6 +26,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
+import org.apache.iotdb.db.sync.conf.Constans;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -34,9 +35,9 @@ import org.slf4j.LoggerFactory;
public class FileManagerTest {
- public static final String POST_BACK_DIRECTORY_TEST = "postback" + File.separator;
+ public static final String POST_BACK_DIRECTORY_TEST = Constans.SYNC + File.separator;
public static final String LAST_FILE_INFO_TEST =
- POST_BACK_DIRECTORY_TEST + "lastLocalFileList.txt";
+ POST_BACK_DIRECTORY_TEST + Constans.LAST_LOCAL_FILE_NAME;
public static final String SENDER_FILE_PATH_TEST = POST_BACK_DIRECTORY_TEST + "data";
FileManager manager = FileManager.getInstance();
private static final Logger LOGGER = LoggerFactory.getLogger(FileManagerTest.class);
@@ -180,7 +181,7 @@ public class FileManagerTest {
}
String rand = String.valueOf(r.nextInt(10000));
String fileName =
- SENDER_FILE_PATH_TEST + File.separator + String.valueOf(i) + File.separator + rand;
+ SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand;
File file = new File(fileName);
allFileList.get(String.valueOf(i)).add(file.getAbsolutePath());
if (!file.getParentFile().exists()) {
@@ -268,7 +269,7 @@ public class FileManagerTest {
allFileList = manager.getCurrentLocalFiles();
manager.getLastLocalFileList(LAST_FILE_INFO_TEST);
lastlocalList = manager.getLastLocalFiles();
- manager.getSendingFileList();
+ manager.getValidFileList();
assert (lastlocalList.isEmpty());
assert (isEmpty(allFileList));
@@ -302,8 +303,8 @@ public class FileManagerTest {
allFileList = manager.getCurrentLocalFiles();
manager.backupNowLocalFileInfo(LAST_FILE_INFO_TEST);
manager.getLastLocalFileList(LAST_FILE_INFO_TEST);
- manager.getSendingFileList();
- sendingFileList = manager.getSendingFiles();
+ manager.getValidFileList();
+ sendingFileList = manager.getValidAllFiles();
assert (sendingFileList.size() == newFileList.size());
for (Entry<String, Set<String>> entry : sendingFileList.entrySet()) {
assert (newFileList.containsKey(entry.getKey()));
@@ -340,7 +341,7 @@ public class FileManagerTest {
}
String rand = String.valueOf(r.nextInt(10000));
String fileName =
- SENDER_FILE_PATH_TEST + File.separator + String.valueOf(i) + File.separator + rand;
+ SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand;
File file = new File(fileName);
allFileList.get(String.valueOf(i)).add(file.getAbsolutePath());
newFileList.get(String.valueOf(i)).add(file.getAbsolutePath());
@@ -354,8 +355,8 @@ public class FileManagerTest {
}
manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST});
manager.getLastLocalFileList(LAST_FILE_INFO_TEST);
- manager.getSendingFileList();
- sendingFileList = manager.getSendingFiles();
+ manager.getValidFileList();
+ sendingFileList = manager.getValidAllFiles();
assert (sendingFileList.size() == newFileList.size());
for (Entry<String, Set<String>> entry : sendingFileList.entrySet()) {
assert (newFileList.containsKey(entry.getKey()));
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/postback/sender/IoTDBSingleClientPostBackTest.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/IoTDBSingleClientPostBackTest.java
similarity index 97%
rename from iotdb/src/test/java/org/apache/iotdb/db/postback/sender/IoTDBSingleClientPostBackTest.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/sender/IoTDBSingleClientPostBackTest.java
index d401369..ccdafff 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/postback/sender/IoTDBSingleClientPostBackTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/IoTDBSingleClientPostBackTest.java
@@ -16,11 +16,12 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.sender;
+package org.apache.iotdb.db.sync.sender;
import static org.junit.Assert.fail;
import java.io.File;
+import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
@@ -31,9 +32,10 @@ import java.util.Set;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.exception.SyncConnectionException;
import org.apache.iotdb.db.integration.Constant;
-import org.apache.iotdb.db.postback.conf.PostBackSenderConfig;
-import org.apache.iotdb.db.postback.conf.PostBackSenderDescriptor;
+import org.apache.iotdb.db.sync.conf.SyncSenderConfig;
+import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.jdbc.Config;
@@ -41,7 +43,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * The test is to run a complete postback function Before you run the test, make sure receiver has
+ * The test is to run a complete sync function Before you run the test, make sure receiver has
* been cleaned up and inited.
*/
public class IoTDBSingleClientPostBackTest {
@@ -49,7 +51,7 @@ public class IoTDBSingleClientPostBackTest {
FileSenderImpl fileSenderImpl = FileSenderImpl.getInstance();
private IoTDBConfig conf = IoTDBDescriptor.getInstance().getConfig();
private String serverIpTest = "192.168.130.17";
- private PostBackSenderConfig config = PostBackSenderDescriptor.getInstance().getConfig();
+ private SyncSenderConfig config = SyncSenderDescriptor.getInstance().getConfig();
private Set<String> dataSender = new HashSet<>();
private Set<String> dataReceiver = new HashSet<>();
private boolean success = true;
@@ -143,7 +145,7 @@ public class IoTDBSingleClientPostBackTest {
"insert into root.test.d0(timestamp,s1) values(3000,'1309')",
"insert into root.test.d1.g0(timestamp,s0) values(400,1050)", "merge", "flush",};
private boolean testFlag = Constant.testFlag;
- private static final String POSTBACK = "postback";
+ private static final String POSTBACK = "sync";
private static final Logger logger = LoggerFactory.getLogger(IoTDBSingleClientPostBackTest.class);
public static void main(String[] args) throws Exception {
@@ -199,9 +201,9 @@ public class IoTDBSingleClientPostBackTest {
}
}
- public void testPostback() {
+ public void testPostback() throws IOException, SyncConnectionException {
if (testFlag) {
- // the first time to postback
+ // the first time to sync
logger.debug("It's the first time to post back!");
try {
Thread.sleep(2000);
@@ -220,7 +222,7 @@ public class IoTDBSingleClientPostBackTest {
fail(e.getMessage());
}
- fileSenderImpl.postback();
+ fileSenderImpl.sync();
// Compare data of sender and receiver
dataSender.clear();
@@ -315,7 +317,7 @@ public class IoTDBSingleClientPostBackTest {
return;
}
- // the second time to postback
+ // the second time to sync
logger.debug("It's the second time to post back!");
try {
Thread.sleep(2000);
@@ -343,7 +345,7 @@ public class IoTDBSingleClientPostBackTest {
Thread.currentThread().interrupt();
}
- fileSenderImpl.postback();
+ fileSenderImpl.sync();
// Compare data of sender and receiver
dataSender.clear();
@@ -443,7 +445,7 @@ public class IoTDBSingleClientPostBackTest {
return;
}
- // the third time to postback
+ // the third time to sync
logger.debug("It's the third time to post back!");
try {
Thread.sleep(2000);
@@ -465,7 +467,7 @@ public class IoTDBSingleClientPostBackTest {
Thread.currentThread().interrupt();
}
- fileSenderImpl.postback();
+ fileSenderImpl.sync();
// Compare data of sender and receiver
dataSender.clear();
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/postback/sender/MultipleClientPostBackTest.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientPostBackTest.java
similarity index 99%
rename from iotdb/src/test/java/org/apache/iotdb/db/postback/sender/MultipleClientPostBackTest.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientPostBackTest.java
index 2c30fb0..c988471 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/postback/sender/MultipleClientPostBackTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/sender/MultipleClientPostBackTest.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.sender;
+package org.apache.iotdb.db.sync.sender;
import static org.junit.Assert.fail;
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender1.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender1.java
similarity index 96%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender1.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender1.java
index e5e9664..42f1965 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender1.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender1.java
@@ -16,9 +16,9 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.utils;
+package org.apache.iotdb.db.sync.test;
-import static org.apache.iotdb.db.postback.utils.RandomNum.getRandomInt;
+import static org.apache.iotdb.db.sync.test.RandomNum.getRandomInt;
import java.io.BufferedReader;
import java.io.File;
@@ -37,10 +37,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * The class is to generate data of whole timeseries (simulating jilian scene) to test stability of
- * postback function.
+ * CreateDataSender1 is used to generate data of whole timeseries (simulating jilian scene) to test stability of
+ * sync function.
*
- * @author lta
+ * @author Tianan Li
*/
public class CreateDataSender1 {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender2.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender2.java
similarity index 97%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender2.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender2.java
index 07cb144..f335e5e 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender2.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender2.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.utils;
+package org.apache.iotdb.db.sync.test;
import java.io.BufferedReader;
import java.io.File;
@@ -35,10 +35,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * The class is to generate data of half timeseries (simulating jilian scene) to test stability of
- * postback function.
+ * CreateDataSender2 is used to generate data of half timeseries (simulating jilian scene) to test stability of
+ * sync function.
*
- * @author lta
+ * @author Tianan Li
*/
public class CreateDataSender2 {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender3.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender3.java
similarity index 98%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender3.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender3.java
index e3f05a2..83693ca 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/CreateDataSender3.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/CreateDataSender3.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.utils;
+package org.apache.iotdb.db.sync.test;
import java.io.BufferedReader;
import java.io.File;
@@ -35,10 +35,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * The class is to generate data of another half timeseries (simulating jilian scene) which is
- * different to those in CreateDataSender2 to test stability of postback function.
+ * CreateDataSender3 is used to generate data of another half timeseries (simulating jilian scene) which is
+ * different to those in CreateDataSender2 to test stability of sync function.
*
- * @author lta
+ * @author Tianan Li
*/
public class CreateDataSender3 {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/RandomNum.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/RandomNum.java
similarity index 97%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/utils/RandomNum.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/test/RandomNum.java
index db28f3c..1ffee67 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/RandomNum.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/RandomNum.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.utils;
+package org.apache.iotdb.db.sync.test;
import java.util.Random;
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/Utils.java b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/Utils.java
similarity index 94%
rename from iotdb/src/main/java/org/apache/iotdb/db/postback/utils/Utils.java
rename to iotdb/src/test/java/org/apache/iotdb/db/sync/test/Utils.java
index 3958024..5f968d4 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/utils/Utils.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/sync/test/Utils.java
@@ -16,10 +16,10 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.postback.utils;
+package org.apache.iotdb.db.sync.test;
/**
- * Created by stefanie on 07/08/2017.
+ * * @author Tianan Li
*/
public class Utils {