Merge pull request #15 from chrislusf/master

sync

10 files changed, 60 insertions, 35 deletions

diff --git a/other/java/client/pom.xml b/other/java/client/pom.xml
index c8e90c96a..edfd38a43 100644
--- a/other/java/client/pom.xml
+++ b/other/java/client/pom.xml
@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.6</version>
+    <version>1.4.7</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
diff --git a/other/java/client/pom.xml.deploy b/other/java/client/pom.xml.deploy
index c8e90c96a..edfd38a43 100644
--- a/other/java/client/pom.xml.deploy
+++ b/other/java/client/pom.xml.deploy
@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.6</version>
+    <version>1.4.7</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
diff --git a/other/java/client/pom_debug.xml b/other/java/client/pom_debug.xml
index 395efa984..6b553c831 100644
--- a/other/java/client/pom_debug.xml
+++ b/other/java/client/pom_debug.xml
@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.6</version>
+    <version>1.4.7</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>
diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto
index 65947e674..cf88065ef 100644
--- a/other/java/client/src/main/proto/filer.proto
+++ b/other/java/client/src/main/proto/filer.proto
@@ -58,6 +58,12 @@ service SeaweedFiler {
     rpc LocateBroker (LocateBrokerRequest) returns (LocateBrokerResponse) {
     }
 
+    rpc KvGet (KvGetRequest) returns (KvGetResponse) {
+    }
+
+    rpc KvPut (KvPutRequest) returns (KvPutResponse) {
+    }
+
 }
 
 //////////////////////////////////////////////////
@@ -266,6 +272,7 @@ message GetFilerConfigurationResponse {
     uint32 max_mb = 4;
     string dir_buckets = 5;
     bool cipher = 7;
+    int32 signature = 8;
 }
 
 message SubscribeMetadataRequest {
@@ -307,3 +314,19 @@ message LocateBrokerResponse {
     }
     repeated Resource resources = 2;
 }
+
+// Key-Value operations
+message KvGetRequest {
+    bytes key = 1;
+}
+message KvGetResponse {
+    bytes value = 1;
+    string error = 2;
+}
+message KvPutRequest {
+    bytes key = 1;
+    bytes value = 2;
+}
+message KvPutResponse {
+    string error = 1;
+}
diff --git a/other/java/hdfs2/dependency-reduced-pom.xml b/other/java/hdfs2/dependency-reduced-pom.xml
index 1f48d8390..0ba8bbae1 100644
--- a/other/java/hdfs2/dependency-reduced-pom.xml
+++ b/other/java/hdfs2/dependency-reduced-pom.xml
@@ -301,7 +301,7 @@
     </snapshotRepository>

   </distributionManagement>

   <properties>

-    <seaweedfs.client.version>1.4.6</seaweedfs.client.version>

+    <seaweedfs.client.version>1.4.7</seaweedfs.client.version>

     <hadoop.version>2.9.2</hadoop.version>

   </properties>

 </project>

diff --git a/other/java/hdfs2/pom.xml b/other/java/hdfs2/pom.xml
index a8de5bca0..0e6d14332 100644
--- a/other/java/hdfs2/pom.xml
+++ b/other/java/hdfs2/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.4.6</seaweedfs.client.version>
+        <seaweedfs.client.version>1.4.7</seaweedfs.client.version>
         <hadoop.version>2.9.2</hadoop.version>
     </properties>
 
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(
diff --git a/other/java/hdfs3/dependency-reduced-pom.xml b/other/java/hdfs3/dependency-reduced-pom.xml
index 25de9dfc0..691be547e 100644
--- a/other/java/hdfs3/dependency-reduced-pom.xml
+++ b/other/java/hdfs3/dependency-reduced-pom.xml
@@ -309,7 +309,7 @@
     </snapshotRepository>

   </distributionManagement>

   <properties>

-    <seaweedfs.client.version>1.4.6</seaweedfs.client.version>

+    <seaweedfs.client.version>1.4.7</seaweedfs.client.version>

     <hadoop.version>3.1.1</hadoop.version>

   </properties>

 </project>

diff --git a/other/java/hdfs3/pom.xml b/other/java/hdfs3/pom.xml
index afb6ef309..8e8ec1958 100644
--- a/other/java/hdfs3/pom.xml
+++ b/other/java/hdfs3/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.4.6</seaweedfs.client.version>
+        <seaweedfs.client.version>1.4.7</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 
diff --git a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
index fd8877806..6551548fa 100644
--- a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
+++ b/other/java/hdfs3/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(