Merge pull request #8 from chrislusf/master

Hadoop file system: 1.4.3

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.1</version>
+    <version>1.4.3</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom.xml.deploy

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.1</version>
+    <version>1.4.3</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom_debug.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.4.1</version>
+    <version>1.4.3</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/hdfs2/dependency-reduced-pom.xml

@@ -127,7 +127,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.4.1</seaweedfs.client.version>
+    <seaweedfs.client.version>1.4.3</seaweedfs.client.version>
     <hadoop.version>2.9.2</hadoop.version>
   </properties>
 </project>

other/java/hdfs2/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.4.1</seaweedfs.client.version>
+        <seaweedfs.client.version>1.4.3</seaweedfs.client.version>
         <hadoop.version>2.9.2</hadoop.version>
     </properties>
 

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

@@ -1,49 +0,0 @@
-package seaweed.hdfs;
-
-import org.apache.hadoop.fs.PositionedReadable;
-import org.apache.hadoop.fs.Seekable;
-
-import java.io.BufferedInputStream;
-import java.io.FilterInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-
-class BufferedSeaweedInputStream extends FilterInputStream implements Seekable, PositionedReadable {
-
-    SeaweedInputStream t;
-
-    protected BufferedSeaweedInputStream(InputStream in, int bufferSize) {
-        super(new BufferedInputStream(in, bufferSize));
-        t = (SeaweedInputStream)in;
-    }
-
-    @Override
-    public int read(long position, byte[] buffer, int offset, int length) throws IOException {
-        return this.t.read(position,buffer,offset,length);
-    }
-
-    @Override
-    public void readFully(long position, byte[] buffer, int offset, int length) throws IOException {
-        this.t.readFully(position,buffer,offset,length);
-    }
-
-    @Override
-    public void readFully(long position, byte[] buffer) throws IOException {
-        this.t.readFully(position,buffer);
-    }
-
-    @Override
-    public void seek(long pos) throws IOException {
-        this.t.seek(pos);
-    }
-
-    @Override
-    public long getPos() throws IOException {
-        return this.t.getPos();
-    }
-
-    @Override
-    public boolean seekToNewSource(long targetPos) throws IOException {
-        return this.t.seekToNewSource(targetPos);
-    }
-}

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

@@ -75,8 +75,8 @@ public class SeaweedFileSystem extends FileSystem {
         path = qualify(path);
 
         try {
-            InputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics, bufferSize);
-            return new FSDataInputStream(new BufferedSeaweedInputStream(inputStream, 16 * 1024 * 1024));
+            FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics, bufferSize);
+            return new FSDataInputStream(new BufferedFSInputStream(inputStream, 16 * 1024 * 1024));
         } catch (Exception ex) {
             LOG.warn("open path: {} bufferSize:{}", path, bufferSize, ex);
             return null;

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

@@ -1,5 +1,6 @@
 package seaweed.hdfs;
 
+import org.apache.hadoop.fs.FSInputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -207,7 +208,7 @@ public class SeaweedFileSystemStore {
 
     }
 
-    public InputStream openFileForRead(final Path path, FileSystem.Statistics statistics,
+    public FSInputStream openFileForRead(final Path path, FileSystem.Statistics statistics,
                                          int bufferSize) throws IOException {
 
         LOG.debug("openFileForRead path:{} bufferSize:{}", path, bufferSize);

other/java/hdfs3/dependency-reduced-pom.xml

@@ -127,7 +127,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.4.1</seaweedfs.client.version>
+    <seaweedfs.client.version>1.4.3</seaweedfs.client.version>
     <hadoop.version>3.1.1</hadoop.version>
   </properties>
 </project>

other/java/hdfs3/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.4.1</seaweedfs.client.version>
+        <seaweedfs.client.version>1.4.3</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 

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

@@ -1,49 +0,0 @@
-package seaweed.hdfs;
-
-import org.apache.hadoop.fs.PositionedReadable;
-import org.apache.hadoop.fs.Seekable;
-
-import java.io.BufferedInputStream;
-import java.io.FilterInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-
-class BufferedSeaweedInputStream extends FilterInputStream implements Seekable, PositionedReadable {
-
-    SeaweedInputStream t;
-
-    protected BufferedSeaweedInputStream(InputStream in, int bufferSize) {
-        super(new BufferedInputStream(in, bufferSize));
-        t = (SeaweedInputStream)in;
-    }
-
-    @Override
-    public int read(long position, byte[] buffer, int offset, int length) throws IOException {
-        return this.t.read(position,buffer,offset,length);
-    }
-
-    @Override
-    public void readFully(long position, byte[] buffer, int offset, int length) throws IOException {
-        this.t.readFully(position,buffer,offset,length);
-    }
-
-    @Override
-    public void readFully(long position, byte[] buffer) throws IOException {
-        this.t.readFully(position,buffer);
-    }
-
-    @Override
-    public void seek(long pos) throws IOException {
-        this.t.seek(pos);
-    }
-
-    @Override
-    public long getPos() throws IOException {
-        return this.t.getPos();
-    }
-
-    @Override
-    public boolean seekToNewSource(long targetPos) throws IOException {
-        return this.t.seekToNewSource(targetPos);
-    }
-}

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

@@ -75,8 +75,8 @@ public class SeaweedFileSystem extends FileSystem {
         path = qualify(path);
 
         try {
-            InputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics, bufferSize);
-            return new FSDataInputStream(new BufferedSeaweedInputStream(inputStream, 16 * 1024 * 1024));
+            FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics, bufferSize);
+            return new FSDataInputStream(new BufferedFSInputStream(inputStream, 16 * 1024 * 1024));
         } catch (Exception ex) {
             LOG.warn("open path: {} bufferSize:{}", path, bufferSize, ex);
             return null;

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

@@ -1,5 +1,6 @@
 package seaweed.hdfs;
 
+import org.apache.hadoop.fs.FSInputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -207,7 +208,7 @@ public class SeaweedFileSystemStore {
 
     }
 
-    public InputStream openFileForRead(final Path path, FileSystem.Statistics statistics,
+    public FSInputStream openFileForRead(final Path path, FileSystem.Statistics statistics,
                                          int bufferSize) throws IOException {
 
         LOG.debug("openFileForRead path:{} bufferSize:{}", path, bufferSize);