Merge pull request #1621 from hilimd/master

HCFS: add replication parameter and fix close sequence

other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java

@@ -25,6 +25,7 @@ public class SeaweedFileSystem extends FileSystem {
     public static final String FS_SEAWEED_FILER_PORT = "fs.seaweed.filer.port";
     public static final int FS_SEAWEED_DEFAULT_PORT = 8888;
     public static final String FS_SEAWEED_BUFFER_SIZE = "fs.seaweed.buffer.size";
+    public static final String FS_SEAWEED_REPLICATION = "fs.seaweed.replication";
     public static final int FS_SEAWEED_DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024;
 
     private static final Logger LOG = LoggerFactory.getLogger(SeaweedFileSystem.class);
@@ -67,8 +68,8 @@ public class SeaweedFileSystem extends FileSystem {
 
     @Override
     public void close() throws IOException {
-        this.seaweedFileSystemStore.close();
         super.close();
+        this.seaweedFileSystemStore.close();
     }
 
     @Override
@@ -97,7 +98,7 @@ public class SeaweedFileSystem extends FileSystem {
         path = qualify(path);
 
         try {
-            String replicaPlacement = String.format("%03d", replication - 1);
+            String replicaPlacement = this.getConf().get(FS_SEAWEED_REPLICATION, String.format("%03d", replication - 1));
             int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
             OutputStream outputStream = seaweedFileSystemStore.createFile(path, overwrite, permission, seaweedBufferSize, replicaPlacement);
             return new FSDataOutputStream(outputStream, statistics);

other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java

@@ -25,6 +25,7 @@ public class SeaweedFileSystem extends FileSystem {
     public static final String FS_SEAWEED_FILER_PORT = "fs.seaweed.filer.port";
     public static final int FS_SEAWEED_DEFAULT_PORT = 8888;
     public static final String FS_SEAWEED_BUFFER_SIZE = "fs.seaweed.buffer.size";
+    public static final String FS_SEAWEED_REPLICATION = "fs.seaweed.replication";
     public static final int FS_SEAWEED_DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024;
 
     private static final Logger LOG = LoggerFactory.getLogger(SeaweedFileSystem.class);
@@ -67,8 +68,8 @@ public class SeaweedFileSystem extends FileSystem {
 
     @Override
     public void close() throws IOException {
-        this.seaweedFileSystemStore.close();
         super.close();
+        this.seaweedFileSystemStore.close();
     }
 
     @Override
@@ -97,7 +98,7 @@ public class SeaweedFileSystem extends FileSystem {
         path = qualify(path);
 
         try {
-            String replicaPlacement = String.format("%03d", replication - 1);
+            String replicaPlacement = this.getConf().get(FS_SEAWEED_REPLICATION, String.format("%03d", replication - 1));
             int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
             OutputStream outputStream = seaweedFileSystemStore.createFile(path, overwrite, permission, seaweedBufferSize, replicaPlacement);
             return new FSDataOutputStream(outputStream, statistics);

other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java

@@ -287,4 +287,4 @@ public class SeaweedFileSystemStore {
 
     }
 
-}
+}