1.4.7 hdfs configurable fs.seaweed.buffer.size

1 files changed, 15 insertions, 14 deletions

diff --git a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
index fd8877806..6551548fa 100644
--- a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
+++ b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
@@ -5,7 +5,6 @@ import org.apache.hadoop.fs.*;
 import org.apache.hadoop.fs.permission.AclEntry;
 import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.Progressable;
 import org.slf4j.Logger;
@@ -14,20 +13,19 @@ import seaweedfs.client.FilerProto;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.URI;
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_KEY;
-
 public class SeaweedFileSystem extends FileSystem {
 
-    public static final int FS_SEAWEED_DEFAULT_PORT = 8888;
     public static final String FS_SEAWEED_FILER_HOST = "fs.seaweed.filer.host";
     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 int FS_SEAWEED_DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024;
 
     private static final Logger LOG = LoggerFactory.getLogger(SeaweedFileSystem.class);
 
@@ -75,8 +73,9 @@ public class SeaweedFileSystem extends FileSystem {
         path = qualify(path);
 
         try {
-            FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics, bufferSize);
-            return new FSDataInputStream(new BufferedFSInputStream(inputStream, 16 * 1024 * 1024));
+            int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
+            FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics, seaweedBufferSize);
+            return new FSDataInputStream(new BufferedFSInputStream(inputStream, 4 * seaweedBufferSize));
         } catch (Exception ex) {
             LOG.warn("open path: {} bufferSize:{}", path, bufferSize, ex);
             return null;
@@ -93,7 +92,8 @@ public class SeaweedFileSystem extends FileSystem {
 
         try {
             String replicaPlacement = String.format("%03d", replication - 1);
-            OutputStream outputStream = seaweedFileSystemStore.createFile(path, overwrite, permission, bufferSize, replicaPlacement);
+            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);
         } catch (Exception ex) {
             LOG.warn("create path: {} bufferSize:{} blockSize:{}", path, bufferSize, blockSize, ex);
@@ -103,8 +103,9 @@ public class SeaweedFileSystem extends FileSystem {
 
     /**
      * {@inheritDoc}
+     *
      * @throws FileNotFoundException if the parent directory is not present -or
-     * is not a directory.
+     *                               is not a directory.
      */
     @Override
     public FSDataOutputStream createNonRecursive(Path path,
@@ -121,9 +122,10 @@ public class SeaweedFileSystem extends FileSystem {
                 throw new FileAlreadyExistsException("Not a directory: " + parent);
             }
         }
+        int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
         return create(path, permission,
                 flags.contains(CreateFlag.OVERWRITE), bufferSize,
-                replication, blockSize, progress);
+                replication, seaweedBufferSize, progress);
     }
 
     @Override
@@ -133,7 +135,8 @@ public class SeaweedFileSystem extends FileSystem {
 
         path = qualify(path);
         try {
-            OutputStream outputStream = seaweedFileSystemStore.createFile(path, false, null, bufferSize, "");
+            int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
+            OutputStream outputStream = seaweedFileSystemStore.createFile(path, false, null, seaweedBufferSize, "");
             return new FSDataOutputStream(outputStream, statistics);
         } catch (Exception ex) {
             LOG.warn("append path: {} bufferSize:{}", path, bufferSize, ex);
@@ -338,9 +341,7 @@ public class SeaweedFileSystem extends FileSystem {
 
     @Override
     public void createSymlink(final Path target, final Path link,
-                              final boolean createParent) throws AccessControlException,
-            FileAlreadyExistsException, FileNotFoundException,
-            ParentNotDirectoryException, UnsupportedFileSystemException,
+                              final boolean createParent) throws
             IOException {
         // Supporting filesystems should override this method
         throw new UnsupportedOperationException(