This is an automated email from the ASF dual-hosted git repository. lta pushed a commit to branch reimpl_sync in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 32e2ff8af100a3e41fceff0ecf55f984fdfc8657 Author: lta <[email protected]> AuthorDate: Fri Aug 30 20:22:19 2019 +0800 add sync file manager test --- .../iotdb/db/sync/receiver/load/FileLoader.java | 4 +- .../receiver/recover/SyncReceiverLogAnalyzer.java | 2 +- .../db/sync/receiver/transfer/SyncServiceImpl.java | 3 +- .../db/sync/sender/manage/SyncFileManager.java | 17 +- .../sync/sender/recover/SyncSenderLogAnalyzer.java | 2 +- .../sync/sender/transfer/DataTransferManager.java | 6 +- .../iotdb/db/sync/sender/SingleClientSyncTest.java | 596 --------------------- .../iotdb/db/sync/sender/SyncFileManagerTest.java | 374 ------------- .../db/sync/sender/manage/SyncFileManagerTest.java | 276 ++++++++++ 9 files changed, 297 insertions(+), 983 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java index d834487..19f7909 100644 --- a/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java +++ b/server/src/main/java/org/apache/iotdb/db/sync/receiver/load/FileLoader.java @@ -144,8 +144,8 @@ public class FileLoader implements IFileLoader { public void cleanUp() { try { loadLog.close(); - new File(syncFolderPath, Constans.SYNC_LOG_NAME).deleteOnExit(); - new File(syncFolderPath, Constans.LOAD_LOG_NAME).deleteOnExit(); + new File(syncFolderPath, Constans.SYNC_LOG_NAME).delete(); + new File(syncFolderPath, Constans.LOAD_LOG_NAME).delete(); FileLoaderManager.getInstance().removeFileLoader(senderName); } catch (IOException e) { LOGGER.error("Can not clean up sync resource.", e); diff --git a/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogAnalyzer.java index 4ed7284..53a44dd 100644 --- a/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogAnalyzer.java +++ b/server/src/main/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogAnalyzer.java @@ -66,7 +66,7 @@ public class SyncReceiverLogAnalyzer implements ISyncReceiverLogAnalyzer { private boolean recover(File senderFolder) throws IOException { // check the state if (!new File(senderFolder, Constans.SYNC_LOG_NAME).exists()) { - new File(senderFolder, Constans.LOAD_LOG_NAME).deleteOnExit(); + new File(senderFolder, Constans.LOAD_LOG_NAME).delete(); return true; } if (FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName())) { diff --git a/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java index 8d2d61c..12ab86b 100644 --- a/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/sync/receiver/transfer/SyncServiceImpl.java @@ -129,7 +129,8 @@ public class SyncServiceImpl implements SyncService.Iface { * Init file path and clear data if last sync process failed. */ private void initPath() throws DiskSpaceInsufficientException { - String dataDir = DirectoryManager.getInstance().getNextFolderForSequenceFile(); + String dataDir = new File(DirectoryManager.getInstance().getNextFolderForSequenceFile()) + .getParentFile().getAbsolutePath(); syncFolderPath .set(FilePathUtils.regularizePath(dataDir) + Constans.SYNC_RECEIVER + File.separatorChar + senderName.get()); diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java index 23af20c..89ea93d 100644 --- a/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java +++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java @@ -62,8 +62,8 @@ public class SyncFileManager implements ISyncFileManager { private Map<String, Set<File>> deletedFilesMap; /** - * Key is storage group, value is the valid set of new tsfiles which need to be synced to - * receiver end in the sg. + * Key is storage group, value is the valid set of new tsfiles which need to be synced to receiver + * end in the sg. */ private Map<String, Set<File>> toBeSyncedFilesMap; @@ -79,8 +79,12 @@ public class SyncFileManager implements ISyncFileManager { public void getCurrentLocalFiles(String dataDir) { LOGGER.info("Start to get current local files in data folder {}", dataDir); + currentSealedLocalFilesMap = new HashMap<>(); // get all files in data dir sequence folder Map<String, Set<File>> currentAllLocalFiles = new HashMap<>(); + if (!new File(dataDir + File.separatorChar + IoTDBConstant.SEQUENCE_FLODER_NAME).exists()) { + return; + } File[] allSGFolders = new File( dataDir + File.separatorChar + IoTDBConstant.SEQUENCE_FLODER_NAME) .listFiles(); @@ -93,7 +97,6 @@ public class SyncFileManager implements ISyncFileManager { } // get sealed tsfiles - currentSealedLocalFilesMap = new HashMap<>(); for (Entry<String, Set<File>> entry : currentAllLocalFiles.entrySet()) { String sgName = entry.getKey(); currentSealedLocalFilesMap.putIfAbsent(sgName, new HashSet<>()); @@ -115,13 +118,13 @@ public class SyncFileManager implements ISyncFileManager { LOGGER.info("Start to get last local files from last local file info {}", lastLocalFileInfo.getAbsoluteFile()); lastLocalFilesMap = new HashMap<>(); - if(!lastLocalFileInfo.exists()){ + if (!lastLocalFileInfo.exists()) { return; } try (BufferedReader reader = new BufferedReader(new FileReader(lastLocalFileInfo))) { String fileName; while ((fileName = reader.readLine()) != null) { - String sgName = new File(fileName).getParent(); + String sgName = new File(fileName).getParentFile().getName(); allSG.add(sgName); lastLocalFilesMap.putIfAbsent(sgName, new HashSet<>()); lastLocalFilesMap.get(sgName).add(new File(fileName)); @@ -152,6 +155,10 @@ public class SyncFileManager implements ISyncFileManager { } } + public Map<String, Set<File>> getCurrentSealedLocalFilesMap() { + return currentSealedLocalFilesMap; + } + public Map<String, Set<File>> getLastLocalFilesMap() { return lastLocalFilesMap; } diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/recover/SyncSenderLogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/recover/SyncSenderLogAnalyzer.java index 3acfd66..60d45ed 100644 --- a/server/src/main/java/org/apache/iotdb/db/sync/sender/recover/SyncSenderLogAnalyzer.java +++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/recover/SyncSenderLogAnalyzer.java @@ -109,7 +109,7 @@ public class SyncSenderLogAnalyzer implements ISyncSenderLogAnalyzer { } catch (IOException e) { LOGGER.error("Can not clear sync log {}", syncLogFile.getAbsoluteFile(), e); } - lastLocalFile.deleteOnExit(); + lastLocalFile.delete(); currentLocalFile.renameTo(lastLocalFile); } } diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/DataTransferManager.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/DataTransferManager.java index ec75c0a..4969e19 100644 --- a/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/DataTransferManager.java +++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/DataTransferManager.java @@ -512,7 +512,7 @@ public class DataTransferManager implements IDataTransferManager { tsfile, e); } finally { if (snapshotFile != null) { - snapshotFile.deleteOnExit(); + snapshotFile.delete(); } } } @@ -608,7 +608,7 @@ public class DataTransferManager implements IDataTransferManager { } // 2. Rename currentLocalFile to lastLocalFile - lastLocalFile.deleteOnExit(); + lastLocalFile.delete(); currentLocalFile.renameTo(lastLocalFile); // 3. delete snapshot directory @@ -619,7 +619,7 @@ public class DataTransferManager implements IDataTransferManager { } // 4. delete sync log file - getSyncLogFile().deleteOnExit(); + getSyncLogFile().delete(); } diff --git a/server/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java b/server/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java deleted file mode 100644 index 4340670..0000000 --- a/server/src/test/java/org/apache/iotdb/db/sync/sender/SingleClientSyncTest.java +++ /dev/null @@ -1,596 +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.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; -//import java.sql.SQLException; -//import java.sql.Statement; -//import java.util.HashSet; -//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.service.IoTDB; -//import org.apache.iotdb.db.sync.sender.conf.Constans; -//import org.apache.iotdb.db.sync.sender.conf.SyncSenderConfig; -//import org.apache.iotdb.db.sync.sender.conf.SyncSenderDescriptor; -//import org.apache.iotdb.db.sync.sender.transfer.DataTransferManager; -//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. -// */ -//public class SingleClientSyncTest { -// -// DataTransferManager fileSenderImpl = DataTransferManager.getInstance(); -// private IoTDBConfig conf = IoTDBDescriptor.getInstance().getConfig(); -// 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<>(); -// private boolean success = true; -// private IoTDB deamon; -// private static final String[] sqls1 = new String[]{"SET STORAGE GROUP TO root.vehicle", -// "SET STORAGE GROUP TO root.test", -// "CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE", -// "CREATE TIMESERIES root.vehicle.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN", -// "CREATE TIMESERIES root.vehicle.d1.s2 WITH DATATYPE=FLOAT, ENCODING=RLE", -// "CREATE TIMESERIES root.vehicle.d1.s3 WITH DATATYPE=BOOLEAN, ENCODING=PLAIN", -// "CREATE TIMESERIES root.test.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE", -// "CREATE TIMESERIES root.test.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN", -// "CREATE TIMESERIES root.test.d1.g0.s0 WITH DATATYPE=INT32, ENCODING=RLE", -// "insert into root.vehicle.d0(timestamp,s0) values(10,100)", -// "insert into root.vehicle.d0(timestamp,s0,s1) values(12,101,'102')", -// "insert into root.vehicle.d0(timestamp,s1) values(19,'103')", -// "insert into root.vehicle.d1(timestamp,s2) values(11,104.0)", -// "insert into root.vehicle.d1(timestamp,s2,s3) values(15,105.0,true)", -// "insert into root.vehicle.d1(timestamp,s3) values(17,false)", -// "insert into root.vehicle.d0(timestamp,s0) values(20,1000)", -// "insert into root.vehicle.d0(timestamp,s0,s1) values(22,1001,'1002')", -// "insert into root.vehicle.d0(timestamp,s1) values(29,'1003')", -// "insert into root.vehicle.d1(timestamp,s2) values(21,1004.0)", -// "insert into root.vehicle.d1(timestamp,s2,s3) values(25,1005.0,true)", -// "insert into root.vehicle.d1(timestamp,s3) values(27,true)", -// "insert into root.test.d0(timestamp,s0) values(10,106)", -// "insert into root.test.d0(timestamp,s0,s1) values(14,107,'108')", -// "insert into root.test.d0(timestamp,s1) values(16,'109')", -// "insert into root.test.d1.g0(timestamp,s0) values(1,110)", -// "insert into root.test.d0(timestamp,s0) values(30,1006)", -// "insert into root.test.d0(timestamp,s0,s1) values(34,1007,'1008')", -// "insert into root.test.d0(timestamp,s1) values(36,'1090')", -// "insert into root.test.d1.g0(timestamp,s0) values(10,1100)", "merge", "flush",}; -// private static final String[] sqls2 = new String[]{ -// "insert into root.vehicle.d0(timestamp,s0) values(6,120)", -// "insert into root.vehicle.d0(timestamp,s0,s1) values(38,121,'122')", -// "insert into root.vehicle.d0(timestamp,s1) values(9,'123')", -// "insert into root.vehicle.d0(timestamp,s0) values(16,128)", -// "insert into root.vehicle.d0(timestamp,s0,s1) values(18,189,'198')", -// "insert into root.vehicle.d0(timestamp,s1) values(99,'1234')", -// "insert into root.vehicle.d1(timestamp,s2) values(14,1024.0)", -// "insert into root.vehicle.d1(timestamp,s2,s3) values(29,1205.0,true)", -// "insert into root.vehicle.d1(timestamp,s3) values(33,true)", -// "insert into root.test.d0(timestamp,s0) values(15,126)", -// "insert into root.test.d0(timestamp,s0,s1) values(8,127,'128')", -// "insert into root.test.d0(timestamp,s1) values(20,'129')", -// "insert into root.test.d1.g0(timestamp,s0) values(14,430)", -// "insert into root.test.d0(timestamp,s0) values(150,426)", -// "insert into root.test.d0(timestamp,s0,s1) values(80,427,'528')", -// "insert into root.test.d0(timestamp,s1) values(2,'1209')", -// "insert into root.test.d1.g0(timestamp,s0) values(4,330)", "merge", "flush",}; -// private static final String[] sqls3 = new String[]{"SET STORAGE GROUP TO root.iotdb", -// "SET STORAGE GROUP TO root.flush", -// "CREATE TIMESERIES root.iotdb.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE", -// "CREATE TIMESERIES root.iotdb.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN", -// "CREATE TIMESERIES root.iotdb.d1.s2 WITH DATATYPE=FLOAT, ENCODING=RLE", -// "CREATE TIMESERIES root.iotdb.d1.s3 WITH DATATYPE=BOOLEAN, ENCODING=PLAIN", -// "CREATE TIMESERIES root.flush.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE", -// "CREATE TIMESERIES root.flush.d0.s1 WITH DATATYPE=TEXT, ENCODING=PLAIN", -// "CREATE TIMESERIES root.flush.d1.g0.s0 WITH DATATYPE=INT32, ENCODING=RLE", -// "insert into root.iotdb.d0(timestamp,s0) values(3,100)", -// "insert into root.iotdb.d0(timestamp,s0,s1) values(22,101,'102')", -// "insert into root.iotdb.d0(timestamp,s1) values(24,'103')", -// "insert into root.iotdb.d1(timestamp,s2) values(21,104.0)", -// "insert into root.iotdb.d1(timestamp,s2,s3) values(25,105.0,true)", -// "insert into root.iotdb.d1(timestamp,s3) values(27,false)", -// "insert into root.iotdb.d0(timestamp,s0) values(30,1000)", -// "insert into root.iotdb.d0(timestamp,s0,s1) values(202,101,'102')", -// "insert into root.iotdb.d0(timestamp,s1) values(44,'103')", -// "insert into root.iotdb.d1(timestamp,s2) values(1,404.0)", -// "insert into root.iotdb.d1(timestamp,s2,s3) values(250,10.0,true)", -// "insert into root.iotdb.d1(timestamp,s3) values(207,false)", -// "insert into root.flush.d0(timestamp,s0) values(20,106)", -// "insert into root.flush.d0(timestamp,s0,s1) values(14,107,'108')", -// "insert into root.flush.d1.g0(timestamp,s0) values(1,110)", -// "insert into root.flush.d0(timestamp,s0) values(200,1006)", -// "insert into root.flush.d0(timestamp,s0,s1) values(1004,1007,'1080')", -// "insert into root.flush.d1.g0(timestamp,s0) values(1000,910)", -// "insert into root.vehicle.d0(timestamp,s0) values(209,130)", -// "insert into root.vehicle.d0(timestamp,s0,s1) values(206,131,'132')", -// "insert into root.vehicle.d0(timestamp,s1) values(70,'33')", -// "insert into root.vehicle.d1(timestamp,s2) values(204,14.0)", -// "insert into root.vehicle.d1(timestamp,s2,s3) values(29,135.0,false)", -// "insert into root.vehicle.d1(timestamp,s3) values(14,false)", -// "insert into root.test.d0(timestamp,s0) values(19,136)", -// "insert into root.test.d0(timestamp,s0,s1) values(7,137,'138')", -// "insert into root.test.d0(timestamp,s1) values(30,'139')", -// "insert into root.test.d1.g0(timestamp,s0) values(4,150)", -// "insert into root.test.d0(timestamp,s0) values(1900,1316)", -// "insert into root.test.d0(timestamp,s0,s1) values(700,1307,'1038')", -// "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 SYNC_CLIENT = Constans.SYNC_SENDER; -// private static final Logger logger = LoggerFactory.getLogger(SingleClientSyncTest.class); -// -// public static void main(String[] args) throws Exception { -// SingleClientSyncTest singleClientPostBackTest = new SingleClientSyncTest(); -// singleClientPostBackTest.setUp(); -// singleClientPostBackTest.testPostback(); -// singleClientPostBackTest.tearDown(); -// System.exit(0); -// } -// -// public void setConfig() { -// config.setSenderFolderPath( -// config.getDataDirectory() + SYNC_CLIENT + File.separator + Constans.UUID_FILE_NAME); -// config.setLastFileInfo( -// config.getDataDirectory() + SYNC_CLIENT + File.separator + Constans.LAST_LOCAL_FILE_NAME); -// String[] sequenceFileDirectory = config.getSeqFileDirectory(); -// String[] snapshots = new String[config.getSeqFileDirectory().length]; -// for (int i = 0; i < config.getSeqFileDirectory().length; i++) { -// sequenceFileDirectory[i] = new File(sequenceFileDirectory[i]).getAbsolutePath(); -// if (!sequenceFileDirectory[i].endsWith(File.separator)) { -// sequenceFileDirectory[i] = sequenceFileDirectory[i] + File.separator; -// } -// snapshots[i] = -// sequenceFileDirectory[i] + SYNC_CLIENT + File.separator + Constans.DATA_SNAPSHOT_NAME -// + File.separator; -// } -// config.setSnapshotPaths(snapshots); -// config.setSeqFileDirectory(sequenceFileDirectory); -// config.setServerIp(serverIpTest); -// fileSenderImpl.setConfig(config); -// } -// -// public void setUp() throws StartupException, IOException { -// if (testFlag) { -// EnvironmentUtils.closeStatMonitor(); -// deamon = IoTDB.getInstance(); -// deamon.active(); -// EnvironmentUtils.envSetUp(); -// } -// setConfig(); -// } -// -// public void tearDown() throws Exception { -// if (testFlag) { -// deamon.stop(); -// EnvironmentUtils.cleanEnv(); -// } -// if (success) { -// logger.debug("Test succeed!"); -// } else { -// logger.debug("Test failed!"); -// } -// } -// -// public void testPostback() throws IOException, SyncConnectionException, ClassNotFoundException, SQLException, InterruptedException { -// if (testFlag) { -// // the first time to sync -// logger.debug("It's the first time to sync!"); -// try { -// Class.forName(Config.JDBC_DRIVER_NAME); -// try (Connection connection = DriverManager -// .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", -// "root")) { -// Statement statement = connection.createStatement(); -// for (String sql : sqls1) { -// statement.execute(sql); -// } -// statement.close(); -// } -// } catch (SQLException | ClassNotFoundException e) { -// fail(e.getMessage()); -// } -// -// fileSenderImpl.sync(); -// -// // Compare data of sender and receiver -// dataSender.clear(); -// try { -// Class.forName(Config.JDBC_DRIVER_NAME); -// try (Connection connection = DriverManager -// .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", -// "root")) { -// Statement statement = connection.createStatement(); -// boolean hasResultSet = statement.execute("select * from root.vehicle"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataSender.add(res.getString("Time") + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// } -// } -// hasResultSet = statement.execute("select * from root.test"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataSender.add(res.getString("Time") + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// } -// } -// statement.close(); -// } catch (Exception e) { -// logger.error("", e); -// } -// } catch (ClassNotFoundException e) { -// fail(e.getMessage()); -// Thread.currentThread().interrupt(); -// } -// -// dataReceiver.clear(); -// try { -// Class.forName(Config.JDBC_DRIVER_NAME); -// Connection connection = null; -// try { -// connection = DriverManager -// .getConnection(String.format("jdbc:iotdb://%s:6667/", serverIpTest), "root", -// "root"); -// Statement statement = connection.createStatement(); -// boolean hasResultSet = statement.execute("select * from root.vehicle"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataReceiver.add(res.getString("Time") + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// } -// } -// -// hasResultSet = statement.execute("select * from root.test"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataReceiver.add(res.getString("Time") + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// } -// } -// statement.close(); -// } catch (Exception e) { -// logger.error("", e); -// } finally { -// if (connection != null) { -// connection.close(); -// } -// } -// } catch (ClassNotFoundException | SQLException e) { -// fail(e.getMessage()); -// Thread.currentThread().interrupt(); -// } -// logger.debug(String.valueOf(dataSender.size())); -// logger.debug(String.valueOf(dataReceiver.size())); -// logger.debug(dataSender.toString()); -// logger.debug(dataReceiver.toString()); -// if (!(dataSender.size() == dataReceiver.size() && dataSender.containsAll(dataReceiver))) { -// success = false; -// return; -// } -// -// // the second time to sync -// logger.debug("It's the second time to sync!"); -// try { -// Class.forName(Config.JDBC_DRIVER_NAME); -// Connection connection = null; -// try { -// connection = DriverManager -// .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", -// "root"); -// Statement statement = connection.createStatement(); -// for (String sql : sqls2) { -// statement.execute(sql); -// } -// statement.close(); -// } catch (Exception e) { -// logger.error("", e); -// } finally { -// if (connection != null) { -// connection.close(); -// } -// } -// } catch (ClassNotFoundException | SQLException e) { -// fail(e.getMessage()); -// Thread.currentThread().interrupt(); -// } -// -// fileSenderImpl.sync(); -// -// // Compare data of sender and receiver -// dataSender.clear(); -// try { -// Class.forName(Config.JDBC_DRIVER_NAME); -// Connection connection = null; -// try { -// connection = DriverManager -// .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", -// "root"); -// Statement statement = connection.createStatement(); -// boolean hasResultSet = statement.execute("select * from root.vehicle"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataSender.add(res.getString("Time") + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// } -// } -// hasResultSet = statement.execute("select * from root.test"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataSender.add(res.getString("Time") + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// } -// } -// statement.close(); -// } catch (Exception e) { -// logger.error("", e); -// } finally { -// if (connection != null) { -// connection.close(); -// } -// } -// } catch (ClassNotFoundException | SQLException e) { -// fail(e.getMessage()); -// Thread.currentThread().interrupt(); -// } -// -// dataReceiver.clear(); -// { -// Class.forName(Config.JDBC_DRIVER_NAME); -// Connection connection = null; -// try { -// connection = DriverManager -// .getConnection(String.format("jdbc:iotdb://%s:6667/", serverIpTest), "root", -// "root"); -// Statement statement = connection.createStatement(); -// boolean hasResultSet = statement.execute("select * from root.vehicle"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataReceiver.add(res.getString("Time") + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// } -// } -// hasResultSet = statement.execute("select * from root.test"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataReceiver.add(res.getString("Time") + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// } -// } -// statement.close(); -// } catch (Exception e) { -// logger.error("", e); -// } finally { -// if (connection != null) { -// connection.close(); -// } -// } -// } -// logger.debug(String.valueOf(dataSender.size())); -// logger.debug(String.valueOf(dataReceiver.size())); -// logger.debug(dataSender.toString()); -// logger.debug(dataReceiver.toString()); -// if (!(dataSender.size() == dataReceiver.size() && dataSender.containsAll(dataReceiver))) { -// success = false; -// return; -// } -// -// // the third time to sync -// logger.debug("It's the third time to sync!"); -// try { -// Class.forName(Config.JDBC_DRIVER_NAME); -// try (Connection connection = DriverManager -// .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", -// "root")) { -// Statement statement = connection.createStatement(); -// for (String sql : sqls3) { -// statement.execute(sql); -// } -// statement.close(); -// } catch (Exception e) { -// logger.error("", e); -// } -// } catch (ClassNotFoundException e) { -// fail(e.getMessage()); -// Thread.currentThread().interrupt(); -// } -// -// fileSenderImpl.sync(); -// -// // Compare data of sender and receiver -// dataSender.clear(); -// try { -// Class.forName(Config.JDBC_DRIVER_NAME); -// try (Connection connection = DriverManager -// .getConnection(String.format("jdbc:iotdb://%s:6667/", serverIpTest), "root", -// "root")) { -// Statement statement = connection.createStatement(); -// boolean hasResultSet = statement.execute("select * from root.vehicle"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataSender.add(res.getString("Time") + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// } -// } -// hasResultSet = statement.execute("select * from root.test"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataSender.add(res.getString("Time") + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// } -// } -// hasResultSet = statement.execute("select * from root.flush"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataSender.add(res.getString("Time") + res.getString("root.flush.d0.s0") -// + res.getString("root.flush.d0.s1") + res.getString("root.flush.d1.g0.s0")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.flush.d0.s0") -// + res.getString("root.flush.d0.s1") + res.getString("root.flush.d1.g0.s0")); -// } -// } -// hasResultSet = statement.execute("select * from root.iotdb"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataSender.add(res.getString("Time") + res.getString("root.iotdb.d0.s0") -// + res.getString("root.iotdb.d0.s1") + res.getString("root.iotdb.d1.s2") -// + res.getString("root.iotdb.d1.s3")); -// logger.debug(res.getString("Time") + res.getString("root.iotdb.d0.s0") -// + res.getString("root.iotdb.d0.s1") + res.getString("root.iotdb.d1.s2") -// + res.getString("root.iotdb.d1.s3")); -// } -// } -// statement.close(); -// } catch (Exception e) { -// logger.error("", e); -// } -// } catch (ClassNotFoundException e) { -// fail(e.getMessage()); -// Thread.currentThread().interrupt(); -// } -// -// dataReceiver.clear(); -// try { -// Class.forName(Config.JDBC_DRIVER_NAME); -// Connection connection = null; -// try { -// connection = DriverManager -// .getConnection("jdbc:iotdb://192.168.130.8:6667/", "root", "root"); -// Statement statement = connection.createStatement(); -// boolean hasResultSet = statement.execute("select * from root.vehicle"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataReceiver.add(res.getString("Time") + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.vehicle.d0.s0") -// + res.getString("root.vehicle.d0.s1") + res.getString("root.vehicle.d1.s2") -// + res.getString("root.vehicle.d1.s3")); -// } -// } -// hasResultSet = statement.execute("select * from root.test"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataReceiver.add(res.getString("Time") + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.test.d0.s0") -// + res.getString("root.test.d0.s1") + res.getString("root.test.d1.g0.s0")); -// } -// } -// hasResultSet = statement.execute("select * from root.flush"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataReceiver.add(res.getString("Time") + res.getString("root.flush.d0.s0") -// + res.getString("root.flush.d0.s1") + res.getString("root.flush.d1.g0.s0")); -// logger.debug(res.getString("Time") + " | " + res.getString("root.flush.d0.s0") -// + res.getString("root.flush.d0.s1") + res.getString("root.flush.d1.g0.s0")); -// } -// } -// hasResultSet = statement.execute("select * from root.iotdb"); -// if (hasResultSet) { -// ResultSet res = statement.getResultSet(); -// while (res.next()) { -// dataReceiver.add(res.getString("Time") + res.getString("root.iotdb.d0.s0") -// + res.getString("root.iotdb.d0.s1") + res.getString("root.iotdb.d1.s2") -// + res.getString("root.iotdb.d1.s3")); -// logger.debug(res.getString("Time") + res.getString("root.iotdb.d0.s0") -// + res.getString("root.iotdb.d0.s1") + res.getString("root.iotdb.d1.s2") -// + res.getString("root.iotdb.d1.s3")); -// } -// } -// statement.close(); -// } catch (Exception e) { -// logger.error("", e); -// } finally { -// if (connection != null) { -// connection.close(); -// } -// } -// } catch (ClassNotFoundException | SQLException e) { -// fail(e.getMessage()); -// Thread.currentThread().interrupt(); -// } -// logger.debug(String.valueOf(dataSender.size())); -// logger.debug(String.valueOf(dataReceiver.size())); -// logger.debug(String.valueOf(dataSender)); -// logger.debug(String.valueOf(dataReceiver)); -// if (!(dataSender.size() == dataReceiver.size() && dataSender.containsAll(dataReceiver))) { -// success = false; -// } -// } -// } -//} \ No newline at end of file diff --git a/server/src/test/java/org/apache/iotdb/db/sync/sender/SyncFileManagerTest.java b/server/src/test/java/org/apache/iotdb/db/sync/sender/SyncFileManagerTest.java deleted file mode 100644 index 354ef74..0000000 --- a/server/src/test/java/org/apache/iotdb/db/sync/sender/SyncFileManagerTest.java +++ /dev/null @@ -1,374 +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.sync.sender; -// -//import java.io.File; -//import java.io.IOException; -//import java.util.HashMap; -//import java.util.HashSet; -//import java.util.Map; -//import java.util.Map.Entry; -//import java.util.Random; -//import java.util.Set; -//import org.apache.iotdb.db.sync.sender.conf.Constans; -//import org.junit.After; -//import org.junit.Before; -//import org.junit.Test; -//import org.slf4j.Logger; -//import org.slf4j.LoggerFactory; -// -//public class SyncFileManagerTest { -// -// private static final String POST_BACK_DIRECTORY_TEST = Constans.SYNC_SENDER + File.separator; -// private static final String LAST_FILE_INFO_TEST = -// POST_BACK_DIRECTORY_TEST + Constans.LAST_LOCAL_FILE_NAME; -// private static final String SENDER_FILE_PATH_TEST = POST_BACK_DIRECTORY_TEST + "data"; -// private SyncFileManager manager = SyncFileManager.getInstance(); -// private static final Logger logger = LoggerFactory.getLogger(SyncFileManagerTest.class); -// -// @Before -// public void setUp() throws IOException, InterruptedException { -// File file = new File(LAST_FILE_INFO_TEST); -// if (!file.getParentFile().exists()) { -// file.getParentFile().mkdirs(); -// } -// if (!file.exists() && !file.createNewFile()) { -// logger.error("Can not create new file {}", file.getPath()); -// } -// file = new File(SENDER_FILE_PATH_TEST); -// if (!file.exists()) { -// file.mkdirs(); -// } -// manager.setCurrentLocalFiles(new HashMap<>()); -// } -// -// @After -// public void tearDown() throws InterruptedException { -// delete(new File(POST_BACK_DIRECTORY_TEST)); -// new File(POST_BACK_DIRECTORY_TEST).delete(); -// } -// -// public void delete(File file) { -// if (file.isFile() || file.list().length == 0) { -// file.delete(); -// } else { -// File[] files = file.listFiles(); -// for (File f : files) { -// delete(f); -// f.delete(); -// } -// } -// } -// -// @Test // It tests two classes : backupNowLocalFileInfo and getLastLocalFileList -// public void testBackupCurrentLocalFileInfo() throws IOException { -// Map<String, Set<String>> allFileList = new HashMap<>(); -// -// Random r = new Random(0); -// for (int i = 0; i < 3; i++) { -// for (int j = 0; j < 5; j++) { -// if (!allFileList.containsKey(String.valueOf(i))) { -// allFileList.put(String.valueOf(i), new HashSet<>()); -// } -// String rand = String.valueOf(r.nextInt(10000)); -// String fileName = -// SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand; -// File file = new File(fileName); -// allFileList.get(String.valueOf(i)).add(file.getPath()); -// if (!file.getParentFile().exists()) { -// file.getParentFile().mkdirs(); -// } -// if (!file.exists() && !file.createNewFile()) { -// logger.error("Can not create new file {}", file.getPath()); -// } -// } -// } -// Set<String> lastFileList; -// -// // lastFileList is empty -// manager.getLastLocalFileList(LAST_FILE_INFO_TEST); -// lastFileList = manager.getLastLocalFiles(); -// assert (lastFileList.isEmpty()); -// -// // add some files -// manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST}); -// manager.backupNowLocalFileInfo(LAST_FILE_INFO_TEST); -// manager.getLastLocalFileList(LAST_FILE_INFO_TEST); -// lastFileList = manager.getLastLocalFiles(); -// for (Entry<String, Set<String>> entry : allFileList.entrySet()) { -// assert (lastFileList.containsAll(entry.getValue())); -// } -// -// // add some files and delete some files -// r = new Random(1); -// for (int i = 0; i < 3; i++) { -// for (int j = 0; j < 5; j++) { -// if (!allFileList.containsKey(String.valueOf(i))) { -// allFileList.put(String.valueOf(i), new HashSet<>()); -// } -// String rand = String.valueOf(r.nextInt(10000)); -// String fileName = -// SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand; -// File file = new File(fileName); -// allFileList.get(String.valueOf(i)).add(file.getPath()); -// if (!file.getParentFile().exists()) { -// file.getParentFile().mkdirs(); -// } -// if (!file.exists() && !file.createNewFile()) { -// logger.error("Can not create new file {}", file.getPath()); -// } -// } -// } -// int count = 0; -// Map<String, Set<String>> deleteFile = new HashMap<>(); -// for (Entry<String, Set<String>> entry : allFileList.entrySet()) { -// deleteFile.put(entry.getKey(), new HashSet<>()); -// for (String path : entry.getValue()) { -// count++; -// if (count % 3 == 0) { -// deleteFile.get(entry.getKey()).add(path); -// } -// } -// } -// for (Entry<String, Set<String>> entry : deleteFile.entrySet()) { -// for (String path : entry.getValue()) { -// new File(path).delete(); -// allFileList.get(entry.getKey()).remove(path); -// } -// } -// manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST}); -// manager.backupNowLocalFileInfo(LAST_FILE_INFO_TEST); -// manager.getLastLocalFileList(LAST_FILE_INFO_TEST); -// lastFileList = manager.getLastLocalFiles(); -// for (Entry<String, Set<String>> entry : allFileList.entrySet()) { -// assert (lastFileList.containsAll(entry.getValue())); -// } -// } -// -// @Test -// public void testGetCurrentLocalFileList() throws IOException { -// Map<String, Set<String>> allFileList = new HashMap<>(); -// Map<String, Set<String>> fileList; -// -// // nowLocalList is empty -// manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST}); -// fileList = manager.getCurrentLocalFiles(); -// assert (isEmpty(fileList)); -// -// // add some files -// Random r = new Random(0); -// for (int i = 0; i < 3; i++) { -// for (int j = 0; j < 5; j++) { -// if (!allFileList.containsKey(String.valueOf(i))) { -// allFileList.put(String.valueOf(i), new HashSet<>()); -// } -// String rand = String.valueOf(r.nextInt(10000)); -// String fileName = -// SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand; -// File file = new File(fileName); -// allFileList.get(String.valueOf(i)).add(file.getPath()); -// if (!file.getParentFile().exists()) { -// file.getParentFile().mkdirs(); -// } -// if (!file.exists() && !file.createNewFile()) { -// logger.error("Can not create new file {}", file.getPath()); -// } -// } -// } -// manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST}); -// fileList = manager.getCurrentLocalFiles(); -// assert (allFileList.size() == fileList.size()); -// for (Entry<String, Set<String>> entry : fileList.entrySet()) { -// assert (allFileList.containsKey(entry.getKey())); -// assert (allFileList.get(entry.getKey()).containsAll(entry.getValue())); -// } -// -// // delete some files and add some files -// int count = 0; -// Map<String, Set<String>> deleteFile = new HashMap<>(); -// for (Entry<String, Set<String>> entry : allFileList.entrySet()) { -// deleteFile.put(entry.getKey(), new HashSet<>()); -// for (String path : entry.getValue()) { -// count++; -// if (count % 3 == 0) { -// deleteFile.get(entry.getKey()).add(path); -// } -// } -// } -// for (Entry<String, Set<String>> entry : deleteFile.entrySet()) { -// for (String path : entry.getValue()) { -// new File(path).delete(); -// allFileList.get(entry.getKey()).remove(path); -// } -// } -// r = new Random(1); -// for (int i = 0; i < 3; i++) { -// for (int j = 0; j < 5; j++) { -// if (!allFileList.containsKey(String.valueOf(i))) { -// allFileList.put(String.valueOf(i), new HashSet<>()); -// } -// String rand = String.valueOf(r.nextInt(10000)); -// String fileName = -// SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand; -// File file = new File(fileName); -// allFileList.get(String.valueOf(i)).add(file.getPath()); -// if (!file.getParentFile().exists()) { -// file.getParentFile().mkdirs(); -// } -// if (!file.exists() && !file.createNewFile()) { -// logger.error("Can not create new file {}", file.getPath()); -// } -// } -// } -// manager.setCurrentLocalFiles(new HashMap<>()); -// manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST}); -// fileList = manager.getCurrentLocalFiles(); -// assert (allFileList.size() == fileList.size()); -// for (Entry<String, Set<String>> entry : fileList.entrySet()) { -// assert (allFileList.containsKey(entry.getKey())); -// logger.debug("allFileList"); -// for (String a : allFileList.get(entry.getKey())) { -// logger.debug(a); -// } -// logger.debug("FileList"); -// for (String a : entry.getValue()) { -// logger.debug(a); -// } -// assert (allFileList.get(entry.getKey()).containsAll(entry.getValue())); -// } -// } -// -// @Test -// 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; -// -// // nowSendingList is empty -// -// manager.setCurrentLocalFiles(new HashMap<>()); -// manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST}); -// allFileList = manager.getCurrentLocalFiles(); -// manager.getLastLocalFileList(LAST_FILE_INFO_TEST); -// lastLocalList = manager.getLastLocalFiles(); -// manager.getValidFileList(); -// assert (lastLocalList.isEmpty()); -// assert (isEmpty(allFileList)); -// -// // add some files -// newFileList.clear(); -// manager.backupNowLocalFileInfo(LAST_FILE_INFO_TEST); -// Random r = new Random(0); -// for (int i = 0; i < 3; i++) { -// for (int j = 0; j < 5; j++) { -// if (!allFileList.containsKey(String.valueOf(i))) { -// allFileList.put(String.valueOf(i), new HashSet<>()); -// } -// if (!newFileList.containsKey(String.valueOf(i))) { -// newFileList.put(String.valueOf(i), new HashSet<>()); -// } -// String rand = String.valueOf(r.nextInt(10000)); -// String fileName = -// SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand; -// File file = new File(fileName); -// allFileList.get(String.valueOf(i)).add(file.getPath()); -// newFileList.get(String.valueOf(i)).add(file.getPath()); -// if (!file.getParentFile().exists()) { -// file.getParentFile().mkdirs(); -// } -// if (!file.exists() && !file.createNewFile()) { -// logger.error("Can not create new file {}", file.getPath()); -// } -// } -// } -// manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST}); -// allFileList = manager.getCurrentLocalFiles(); -// manager.backupNowLocalFileInfo(LAST_FILE_INFO_TEST); -// manager.getLastLocalFileList(LAST_FILE_INFO_TEST); -// manager.getValidFileList(); -// sendingFileList = manager.getValidAllFiles(); -// assert (sendingFileList.size() == newFileList.size()); -// for (Entry<String, Set<String>> entry : sendingFileList.entrySet()) { -// assert (newFileList.containsKey(entry.getKey())); -// assert (newFileList.get(entry.getKey()).containsAll(entry.getValue())); -// } -// -// // delete some files and add some files -// int count = 0; -// Map<String, Set<String>> deleteFile = new HashMap<>(); -// for (Entry<String, Set<String>> entry : allFileList.entrySet()) { -// deleteFile.put(entry.getKey(), new HashSet<>()); -// for (String path : entry.getValue()) { -// count++; -// if (count % 3 == 0) { -// deleteFile.get(entry.getKey()).add(path); -// } -// } -// } -// for (Entry<String, Set<String>> entry : deleteFile.entrySet()) { -// for (String path : entry.getValue()) { -// new File(path).delete(); -// allFileList.get(entry.getKey()).remove(path); -// } -// } -// newFileList.clear(); -// r = new Random(1); -// for (int i = 0; i < 3; i++) { -// for (int j = 0; j < 5; j++) { -// if (!allFileList.containsKey(String.valueOf(i))) { -// allFileList.put(String.valueOf(i), new HashSet<>()); -// } -// if (!newFileList.containsKey(String.valueOf(i))) { -// newFileList.put(String.valueOf(i), new HashSet<>()); -// } -// String rand = String.valueOf(r.nextInt(10000)); -// String fileName = -// SENDER_FILE_PATH_TEST + File.separator + i + File.separator + rand; -// File file = new File(fileName); -// allFileList.get(String.valueOf(i)).add(file.getPath()); -// newFileList.get(String.valueOf(i)).add(file.getPath()); -// if (!file.getParentFile().exists()) { -// file.getParentFile().mkdirs(); -// } -// if (!file.exists() && !file.createNewFile()) { -// logger.error("Can not create new file {}", file.getPath()); -// } -// } -// } -// manager.getCurrentLocalFileList(new String[]{SENDER_FILE_PATH_TEST}); -// manager.getLastLocalFileList(LAST_FILE_INFO_TEST); -// manager.getValidFileList(); -// sendingFileList = manager.getValidAllFiles(); -// assert (sendingFileList.size() == newFileList.size()); -// for (Entry<String, Set<String>> entry : sendingFileList.entrySet()) { -// assert (newFileList.containsKey(entry.getKey())); -// assert (newFileList.get(entry.getKey()).containsAll(entry.getValue())); -// } -// } -// -// private boolean isEmpty(Map<String, Set<String>> sendingFileList) { -// for (Entry<String, Set<String>> entry : sendingFileList.entrySet()) { -// if (!entry.getValue().isEmpty()) { -// return false; -// } -// } -// return true; -// } -//} \ No newline at end of file diff --git a/server/src/test/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManagerTest.java b/server/src/test/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManagerTest.java new file mode 100644 index 0000000..91e59d3 --- /dev/null +++ b/server/src/test/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManagerTest.java @@ -0,0 +1,276 @@ +package org.apache.iotdb.db.sync.sender.manage; + +import java.io.BufferedWriter; +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Random; +import java.util.Set; +import org.apache.iotdb.db.conf.IoTDBConstant; +import org.apache.iotdb.db.conf.directories.DirectoryManager; +import org.apache.iotdb.db.engine.storagegroup.TsFileResource; +import org.apache.iotdb.db.exception.DiskSpaceInsufficientException; +import org.apache.iotdb.db.exception.StartupException; +import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.sync.sender.conf.SyncSenderConfig; +import org.apache.iotdb.db.sync.sender.conf.SyncSenderDescriptor; +import org.apache.iotdb.db.utils.EnvironmentUtils; +import org.apache.iotdb.db.utils.FilePathUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class SyncFileManagerTest { + + private static final Logger logger = LoggerFactory.getLogger(SyncFileManagerTest.class); + private SyncFileManager manager = SyncFileManager.getInstance(); + private SyncSenderConfig config = SyncSenderDescriptor.getInstance().getConfig(); + private String dataDir; + + @Before + public void setUp() + throws IOException, InterruptedException, StartupException, DiskSpaceInsufficientException { + EnvironmentUtils.envSetUp(); + dataDir = new File(DirectoryManager.getInstance().getNextFolderForSequenceFile()) + .getParentFile().getAbsolutePath(); + config.update(dataDir); + } + + @After + public void tearDown() throws InterruptedException, IOException, StorageEngineException { + EnvironmentUtils.cleanEnv(); + } + + @Test + public void testGetValidFiles() throws IOException { + Map<String, Set<File>> allFileList = new HashMap<>(); + + Random r = new Random(0); + for (int i = 0; i < 3; i++) { + for (int j = 0; j < 5; j++) { + if (!allFileList.containsKey(String.valueOf(i))) { + allFileList.put(String.valueOf(i), new HashSet<>()); + } + String rand = String.valueOf(r.nextInt(10000)); + String fileName = FilePathUtils.regularizePath(dataDir) + IoTDBConstant.SEQUENCE_FLODER_NAME + + File.separator + i + + File.separator + rand; + File file = new File(fileName); + allFileList.get(String.valueOf(i)).add(file); + if (!file.getParentFile().exists()) { + file.getParentFile().mkdirs(); + } + if (!file.exists() && !file.createNewFile()) { + logger.error("Can not create new file {}", file.getPath()); + } + if (!new File(file.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).exists() + && !new File(file.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).createNewFile()) { + logger.error("Can not create new file {}", file.getPath()); + } + } + } + Map<String, Set<File>> lastFileMap; + Map<String, Set<File>> curFileMap; + Map<String, Set<File>> deletedFilesMap; + Map<String, Set<File>> toBeSyncedFilesMap; + + // lastFileList is empty + manager.getValidFiles(dataDir); + assert isEmpty(manager.getLastLocalFilesMap()); + + updateLastLocalFiles(allFileList); + + manager.getValidFiles(dataDir); + lastFileMap = manager.getLastLocalFilesMap(); + for (Entry<String, Set<File>> entry : allFileList.entrySet()) { + assert lastFileMap.containsKey(entry.getKey()); + assert lastFileMap.get(entry.getKey()).containsAll(entry.getValue()); + } + + // add some files + Map<String, Set<File>> correctToBeSyncedFiles = new HashMap<>(); + r = new Random(1); + for (int i = 0; i < 3; i++) { + for (int j = 0; j < 5; j++) { + if (!allFileList.containsKey(String.valueOf(i))) { + allFileList.put(String.valueOf(i), new HashSet<>()); + } + correctToBeSyncedFiles.putIfAbsent(String.valueOf(i), new HashSet<>()); + String rand = String.valueOf(r.nextInt(10000)); + String fileName = + FilePathUtils.regularizePath(dataDir) + IoTDBConstant.SEQUENCE_FLODER_NAME + + File.separator + i + + File.separator + rand; + File file = new File(fileName); + allFileList.get(String.valueOf(i)).add(file); + correctToBeSyncedFiles.get(String.valueOf(i)).add(file); + if (!file.getParentFile().exists()) { + file.getParentFile().mkdirs(); + } + if (!file.exists() && !file.createNewFile()) { + logger.error("Can not create new file {}", file.getPath()); + } + if (!new File(file.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).exists() + && !new File(file.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).createNewFile()) { + logger.error("Can not create new file {}", file.getPath()); + } + } + } + manager.getValidFiles(dataDir); + lastFileMap = manager.getLastLocalFilesMap(); + curFileMap = manager.getCurrentSealedLocalFilesMap(); + deletedFilesMap = manager.getDeletedFilesMap(); + toBeSyncedFilesMap = manager.getToBeSyncedFilesMap(); + for (Entry<String, Set<File>> entry : allFileList.entrySet()) { + assert curFileMap.containsKey(entry.getKey()); + assert curFileMap.get(entry.getKey()).containsAll(entry.getValue()); + } + for (Entry<String, Set<File>> entry : correctToBeSyncedFiles.entrySet()) { + assert toBeSyncedFilesMap.containsKey(entry.getKey()); + assert toBeSyncedFilesMap.get(entry.getKey()).containsAll(entry.getValue()); + } + updateLastLocalFiles(allFileList); + manager.getValidFiles(dataDir); + lastFileMap = manager.getLastLocalFilesMap(); + for (Entry<String, Set<File>> entry : allFileList.entrySet()) { + assert lastFileMap.containsKey(entry.getKey()); + assert lastFileMap.get(entry.getKey()).containsAll(entry.getValue()); + } + + // add some files and delete some files + correctToBeSyncedFiles.clear(); + r = new Random(2); + for (int i = 0; i < 3; i++) { + for (int j = 0; j < 5; j++) { + if (!allFileList.containsKey(String.valueOf(i))) { + allFileList.put(String.valueOf(i), new HashSet<>()); + } + correctToBeSyncedFiles.putIfAbsent(String.valueOf(i), new HashSet<>()); + String rand = String.valueOf(r.nextInt(10000)); + String fileName = + FilePathUtils.regularizePath(dataDir) + IoTDBConstant.SEQUENCE_FLODER_NAME + + File.separator + i + + File.separator + rand; + File file = new File(fileName); + allFileList.get(String.valueOf(i)).add(file); + correctToBeSyncedFiles.get(String.valueOf(i)).add(file); + if (!file.getParentFile().exists()) { + file.getParentFile().mkdirs(); + } + if (!file.exists() && !file.createNewFile()) { + logger.error("Can not create new file {}", file.getPath()); + } + if (!new File(file.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).exists() + && !new File(file.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).createNewFile()) { + logger.error("Can not create new file {}", file.getPath()); + } + } + } + int count = 0; + Map<String, Set<File>> correctDeleteFile = new HashMap<>(); + for (Entry<String, Set<File>> entry : allFileList.entrySet()) { + correctDeleteFile.put(entry.getKey(), new HashSet<>()); + for (File file : entry.getValue()) { + count++; + if (count % 3 == 0 && lastFileMap.get(entry.getKey()).contains(file)) { + correctDeleteFile.get(entry.getKey()).add(file); + } + } + } + for (Entry<String, Set<File>> entry : correctDeleteFile.entrySet()) { + for (File file : entry.getValue()) { + file.delete(); + new File(file.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).delete(); + allFileList.get(entry.getKey()).remove(file); + } + } + manager.getValidFiles(dataDir); + lastFileMap = manager.getLastLocalFilesMap(); + curFileMap = manager.getCurrentSealedLocalFilesMap(); + deletedFilesMap = manager.getDeletedFilesMap(); + toBeSyncedFilesMap = manager.getToBeSyncedFilesMap(); + for (Entry<String, Set<File>> entry : allFileList.entrySet()) { + assert curFileMap.containsKey(entry.getKey()); + assert curFileMap.get(entry.getKey()).containsAll(entry.getValue()); + } + for (Entry<String, Set<File>> entry : correctDeleteFile.entrySet()) { + assert deletedFilesMap.containsKey(entry.getKey()); + assert deletedFilesMap.get(entry.getKey()).containsAll(entry.getValue()); + } + for (Entry<String, Set<File>> entry : correctToBeSyncedFiles.entrySet()) { + assert toBeSyncedFilesMap.containsKey(entry.getKey()); + assert toBeSyncedFilesMap.get(entry.getKey()).containsAll(entry.getValue()); + } + + // add some invalid files + r = new Random(3); + for (int i = 0; i < 3; i++) { + for (int j = 0; j < 5; j++) { + if (!allFileList.containsKey(String.valueOf(i))) { + allFileList.put(String.valueOf(i), new HashSet<>()); + } + String rand = String.valueOf(r.nextInt(10000)); + String fileName = + FilePathUtils.regularizePath(dataDir) + IoTDBConstant.SEQUENCE_FLODER_NAME + + File.separator + i + + File.separator + rand; + File file = new File(fileName); + allFileList.get(String.valueOf(i)).add(file); + if (!file.getParentFile().exists()) { + file.getParentFile().mkdirs(); + } + if (!file.exists() && !file.createNewFile()) { + logger.error("Can not create new file {}", file.getPath()); + } + } + } + manager.getValidFiles(dataDir); + lastFileMap = manager.getLastLocalFilesMap(); + curFileMap = manager.getCurrentSealedLocalFilesMap(); + deletedFilesMap = manager.getDeletedFilesMap(); + toBeSyncedFilesMap = manager.getToBeSyncedFilesMap(); + for (Entry<String, Set<File>> entry : curFileMap.entrySet()) { + assert allFileList.containsKey(entry.getKey()); + assert allFileList.get(entry.getKey()).size() != entry.getValue().size(); + assert allFileList.get(entry.getKey()).containsAll(entry.getValue()); + } + for (Entry<String, Set<File>> entry : correctDeleteFile.entrySet()) { + assert deletedFilesMap.containsKey(entry.getKey()); + assert deletedFilesMap.get(entry.getKey()).containsAll(entry.getValue()); + } + for (Entry<String, Set<File>> entry : correctToBeSyncedFiles.entrySet()) { + assert toBeSyncedFilesMap.containsKey(entry.getKey()); + assert toBeSyncedFilesMap.get(entry.getKey()).containsAll(entry.getValue()); + } + } + + private void updateLastLocalFiles(Map<String, Set<File>> lastLocalFilesMap) { + try (BufferedWriter bw = new BufferedWriter( + new FileWriter(new File(config.getLastFileInfo())))) { + for (Set<File> currentLocalFiles : lastLocalFilesMap.values()) { + for (File file : currentLocalFiles) { + bw.write(file.getAbsolutePath()); + bw.newLine(); + } + bw.flush(); + } + } catch (IOException e) { + logger.error("Can not clear sync log {}", config.getLastFileInfo(), e); + } + } + + private boolean isEmpty(Map<String, Set<File>> sendingFileList) { + for (Entry<String, Set<File>> entry : sendingFileList.entrySet()) { + if (!entry.getValue().isEmpty()) { + return false; + } + } + return true; + } +} \ No newline at end of file
