[FLINK-1081] Add getPos() method into FSDataInputStream class

flink-core/src/main/java/org/apache/flink/api/common/io/InflaterInputStreamFSInputWrapper.java

@@ -37,6 +37,11 @@ public void seek(long desired) throws IOException {
  throw new UnsupportedOperationException("Compressed streams do not support the seek operation");
  }
 
+ @Override
+ public long getPos() throws IOException {
+ throw new UnsupportedOperationException("Compressed streams do not support the getPos operation");
+ }
+
  @Override
  public int read() throws IOException {
  return inStream.read();

flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java

@@ -45,4 +45,10 @@ public abstract class FSDataInputStream extends InputStream {
  */
  public abstract void seek(long desired) throws IOException;
 
+ /**
+ * Get the current position in the input stream.
+ *
+ * @return current position in the input stream
+ */
+ public abstract long getPos() throws IOException;
 }

flink-core/src/main/java/org/apache/flink/core/fs/local/LocalDataInputStream.java

@@ -57,6 +57,11 @@ public void seek(final long desired) throws IOException {
  this.fis.getChannel().position(desired);
  }
 
+ @Override
+ public long getPos() throws IOException {
+ return this.fis.getChannel().position();
+ }
+
 
  @Override
  public int read() throws IOException {

flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopDataInputStream.java

@@ -48,6 +48,12 @@ public synchronized void seek(long desired) throws IOException {
  fsDataInputStream.seek(desired);
  }
 
+ @Override
+ public long getPos() throws IOException {
+
+ return fsDataInputStream.getPos();
+ }
+
  @Override
  public int read() throws IOException {
 

flink-runtime/src/main/java/org/apache/flink/runtime/fs/s3/S3DataInputStream.java

@@ -32,7 +32,6 @@
 /**
  * This class implements an {@link FSDataInputStream} that downloads its data from Amazon S3 in the background.
  * Essentially, this class is just a wrapper to the Amazon AWS SDK.
- * 
  */
 public class S3DataInputStream extends FSDataInputStream {
 
@@ -41,9 +40,19 @@ public class S3DataInputStream extends FSDataInputStream {
  */
  private final InputStream inputStream;
 
+ /**
+ * The current position of input stream.
+ */
+ private long position;
+
+ /**
+ * The marked position.
+ */
+ private long marked;
+
  /**
  * Constructs a new input stream which reads its data from the specified S3 object.
- * 
+ *
  * @param s3Client
  * the S3 client to connect to Amazon S3.
  * @param bucket
@@ -63,6 +72,8 @@ public class S3DataInputStream extends FSDataInputStream {
  }
 
  this.inputStream = s3o.getObjectContent();
+ this.position = 0;
+ this.marked = 0;
  }
 
 
@@ -84,6 +95,7 @@ public void close() throws IOException {
  public void mark(final int readlimit) {
 
  this.inputStream.mark(readlimit);
+ marked = readlimit;
  }
 
 
@@ -97,34 +109,65 @@ public boolean markSupported() {
  @Override
  public int read() throws IOException {
 
- return this.inputStream.read();
+ int read = this.inputStream.read();
+ if (read != -1) {
+ ++position;
+ }
+
+ return read;
  }
 
 
  @Override
  public int read(final byte[] b) throws IOException {
 
- return this.inputStream.read(b);
+ int read = this.inputStream.read(b);
+ if (read > 0) {
+ position += read;
+ }
+
+ return read;
  }
 
 
  @Override
  public int read(final byte[] b, final int off, final int len) throws IOException {
 
- return this.inputStream.read(b, off, len);
+ int read = this.inputStream.read(b, off, len);
+ if (read > 0) {
+ position += read;
+ }
+
+ return read;
  }
 
 
  @Override
  public void reset() throws IOException {
 
  this.inputStream.reset();
+ position = marked;
  }
 
 
  @Override
  public void seek(final long desired) throws IOException {
 
- this.inputStream.skip(desired);
+ skip(desired);
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ long skipped = this.inputStream.skip(n);
+ if (skipped > 0) {
+ position += skipped;
+ }
+
+ return skipped;
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return position;
  }
 }