This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 0c57d25 [IOTDB-222] Fix changing to read-only mode when flushing
Tsfile on HDFS (#397)
0c57d25 is described below
commit 0c57d25b5f27a767ef132847bbe68eb70f7d2dcd
Author: Zesong Sun <[email protected]>
AuthorDate: Thu Sep 19 09:36:08 2019 +0800
[IOTDB-222] Fix changing to read-only mode when flushing Tsfile on HDFS
(#397)
* Fix changing to read-only mode when flushing Tsfile on HDFS
---
.../db/engine/storagegroup/TsFileResource.java | 11 +++--
.../apache/iotdb/tsfile/fileSystem/HDFSFile.java | 27 +++++++-----
.../iotdb/tsfile/fileSystem/TSFileFactory.java | 49 +++++++++++++++++++++-
3 files changed, 69 insertions(+), 18 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index 1e614e3..05d220d 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -26,7 +26,6 @@ import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
@@ -105,8 +104,8 @@ public class TsFileResource {
}
public void serialize() throws IOException {
- try (OutputStream outputStream = new BufferedOutputStream(
- new FileOutputStream(file + RESOURCE_SUFFIX + TEMP_SUFFIX))) {
+ try (OutputStream outputStream =
TSFileFactory.INSTANCE.getBufferedOutputStream(
+ file + RESOURCE_SUFFIX + TEMP_SUFFIX)) {
ReadWriteIOUtils.write(this.startTimeMap.size(), outputStream);
for (Entry<String, Long> entry : this.startTimeMap.entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), outputStream);
@@ -121,12 +120,12 @@ public class TsFileResource {
File src = TSFileFactory.INSTANCE.getFile(file + RESOURCE_SUFFIX +
TEMP_SUFFIX);
File dest = TSFileFactory.INSTANCE.getFile(file + RESOURCE_SUFFIX);
dest.delete();
- FileUtils.moveFile(src, dest);
+ TSFileFactory.INSTANCE.moveFile(src, dest);
}
public void deSerialize() throws IOException {
- try (InputStream inputStream = new BufferedInputStream(
- new FileInputStream(file + RESOURCE_SUFFIX))) {
+ try (InputStream inputStream =
TSFileFactory.INSTANCE.getBufferedInputStream(
+ file + RESOURCE_SUFFIX)) {
int size = ReadWriteIOUtils.readInt(inputStream);
Map<String, Long> startTimes = new HashMap<>();
for (int i = 0; i < size; i++) {
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/fileSystem/HDFSFile.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/fileSystem/HDFSFile.java
index 599365d..d827b7a 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/fileSystem/HDFSFile.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/fileSystem/HDFSFile.java
@@ -20,13 +20,15 @@
package org.apache.iotdb.tsfile.fileSystem;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.*;
import org.apache.iotdb.tsfile.write.TsFileWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.*;
+import java.io.File;
+import java.io.FileFilter;
+import java.io.FilenameFilter;
+import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
@@ -173,7 +175,7 @@ public class HDFSFile extends File {
@Override
public boolean isDirectory() {
try {
- return fs.getFileStatus(hdfsPath).isDirectory();
+ return exists() && fs.getFileStatus(hdfsPath).isDirectory();
} catch (IOException e) {
logger.error("Fail to judge whether {} is a directory. ",
hdfsPath.toUri().toString(), e);
return false;
@@ -207,7 +209,7 @@ public class HDFSFile extends File {
@Override
public int compareTo(File pathname) {
- if(pathname instanceof HDFSFile) {
+ if (pathname instanceof HDFSFile) {
return hdfsPath.toUri().toString().compareTo(pathname.getPath());
} else {
logger.error("File {} is not HDFS file. ", pathname.getPath());
@@ -218,11 +220,21 @@ public class HDFSFile extends File {
@Override
public boolean equals(Object obj) {
if ((obj != null) && (obj instanceof HDFSFile)) {
- return compareTo((HDFSFile)obj) == 0;
+ return compareTo((HDFSFile) obj) == 0;
}
return false;
}
+ @Override
+ public boolean renameTo(File dest) {
+ try {
+ return fs.rename(hdfsPath, new Path(dest.getAbsolutePath()));
+ } catch (IOException e) {
+ logger.error("Failed to rename file {} to {}. ", hdfsPath.toString(),
dest.getName(), e);
+ return false;
+ }
+ }
+
@Override
public String getParent() {
@@ -310,11 +322,6 @@ public class HDFSFile extends File {
}
@Override
- public boolean renameTo(File dest) {
- throw new UnsupportedOperationException("Unsupported operation.");
- }
-
- @Override
public boolean setLastModified(long time) {
throw new UnsupportedOperationException("Unsupported operation.");
}
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/fileSystem/TSFileFactory.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/fileSystem/TSFileFactory.java
index 6c8fd04..5a62802 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/fileSystem/TSFileFactory.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/fileSystem/TSFileFactory.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.tsfile.fileSystem;
+import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -81,7 +82,7 @@ public enum TSFileFactory {
return new BufferedReader(new FileReader(filePath));
}
} catch (IOException e) {
- logger.error("Fail to get buffered reader. ", e);
+ logger.error("Failed to get buffered reader for {}. ", filePath, e);
return null;
}
}
@@ -96,9 +97,53 @@ public enum TSFileFactory {
return new BufferedWriter(new FileWriter(filePath, append));
}
} catch (IOException e) {
- logger.error("Fail to get buffered writer. ", e);
+ logger.error("Failed to get buffered writer for {}. ", filePath, e);
return null;
}
}
+ public BufferedInputStream getBufferedInputStream(String filePath) {
+ try {
+ if (fSType.equals(fSType.HDFS)) {
+ Path path = new Path(filePath);
+ fs = path.getFileSystem(conf);
+ return new BufferedInputStream(fs.open(path));
+ } else {
+ return new BufferedInputStream(new FileInputStream(filePath));
+ }
+ } catch (IOException e) {
+ logger.error("Failed to get buffered input stream for {}. ", filePath,
e);
+ return null;
+ }
+ }
+
+ public BufferedOutputStream getBufferedOutputStream(String filePath) {
+ try {
+ if (fSType.equals(fSType.HDFS)) {
+ Path path = new Path(filePath);
+ fs = path.getFileSystem(conf);
+ return new BufferedOutputStream(fs.create(path));
+ } else {
+ return new BufferedOutputStream(new FileOutputStream(filePath));
+ }
+ } catch (IOException e) {
+ logger.error("Failed to get buffered output stream for {}. ", filePath,
e);
+ return null;
+ }
+ }
+
+ public void moveFile(File srcFile, File destFile) {
+ try {
+ if (fSType.equals(fSType.HDFS)) {
+ boolean rename = srcFile.renameTo(destFile);
+ if (!rename) {
+ logger.error("Failed to rename file from {} to {}. ",
srcFile.getName(), destFile.getName());
+ }
+ } else {
+ FileUtils.moveFile(srcFile, destFile);
+ }
+ } catch (IOException e) {
+ logger.error("Failed to move file from {} to {}. ",
srcFile.getAbsolutePath(), destFile.getAbsolutePath(), e);
+ }
+ }
}
\ No newline at end of file
