Update S3ASeekableStream Implementation to throw when stream is closed

Author: fuatbasik

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ASeekableStream.java

@@ -2,6 +2,7 @@
 
 import java.io.IOException;
 
+import org.apache.hadoop.fs.FSExceptionMessages;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -14,48 +15,46 @@
 public class S3ASeekableStream extends FSInputStream {
 
     private S3SeekableInputStream inputStream;
+    private long lastReadCurrentPos = 0;
     private final String key;
 
     public static final Logger LOG = LoggerFactory.getLogger(S3ASeekableStream.class);
 
-
-    public S3ASeekableStream(String bucket, String key, S3SeekableInputStreamFactory s3SeekableInputStreamFactory)
-            throws IOException {
+    public S3ASeekableStream(String bucket, String key, S3SeekableInputStreamFactory s3SeekableInputStreamFactory) {
         this.inputStream = s3SeekableInputStreamFactory.createStream(S3URI.of(bucket, key));
         this.key = key;
     }
 
     @Override
     public int read() throws IOException {
+        throwIfClosed();
         return inputStream.read();
     }
 
     @Override
     public void seek(long pos) throws IOException {
+        throwIfClosed();
         inputStream.seek(pos);
     }
 
-    @Override
-    public long getPos() throws IOException {
-        return inputStream.getPos();
-    }
 
     @Override
-    public void close() throws IOException {
-        if (inputStream != null) {
-            inputStream.close();
-            inputStream = null;
-            super.close();
+    public synchronized long getPos() {
+        if (!isClosed()) {
+            lastReadCurrentPos = inputStream.getPos();
         }
+        return lastReadCurrentPos;
     }
 
 
     public void readTail(byte[] buf, int off, int n) throws IOException {
+        throwIfClosed();
         inputStream.readTail(buf, off, n);
     }
 
     @Override
     public int read(byte[] buf, int off, int len) throws IOException {
+        throwIfClosed();
         return inputStream.read(buf, off, len);
     }
 
@@ -65,4 +64,22 @@ public boolean seekToNewSource(long l) throws IOException {
         return false;
     }
 
+    @Override
+    public void close() throws IOException {
+        if (inputStream != null) {
+            inputStream.close();
+            inputStream = null;
+            super.close();
+        }
+    }
+
+    protected void throwIfClosed() throws IOException {
+        if (isClosed()) {
+            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
+        }
+    }
+
+    protected boolean isClosed() {
+        return inputStream == null;
+    }
 }