Review comments

Author: Sital Kedia

core/src/main/java/org/apache/spark/io/ReadAheadInputStream.java

@@ -94,8 +94,8 @@ public ReadAheadInputStream(InputStream inputStream, int bufferSizeInBytes, int
     readAheadBuffer.flip();
   }
 
-  private boolean hasRemaining() {
-    if(activeBuffer.remaining() == 0 && readAheadBuffer.remaining() == 0 && endOfStream) {
+  private boolean isEndOfStream() {
+    if(!activeBuffer.hasRemaining() && !readAheadBuffer.hasRemaining() && endOfStream) {
       return true;
     }
     return  false;
@@ -109,13 +109,11 @@ private void readAsync(final ByteBuffer byteBuffer) throws IOException {
     byteBuffer.position(0);
     byteBuffer.flip();
     readInProgress = true;
+    final byte[] arr = byteBuffer.array();
     stateChangeLock.unlock();
     executorService.execute(new Runnable() {
       @Override
       public void run() {
-        stateChangeLock.lock();
-        byte[] arr = byteBuffer.array();
-        stateChangeLock.unlock();
         // Please note that it is safe to release the lock and read into the read ahead buffer
         // because either of following two conditions will hold - 1. The active buffer has
         // data available to read so the reader will not read from the read ahead buffer.
@@ -139,7 +137,7 @@ public void run() {
             endOfStream = true;
           } else if (!handled) {
             readAborted = true;
-            readException = exception;
+            readException = exception != null ? exception: new Exception("Unknown exception in ReadAheadInputStream");
           } else {
             byteBuffer.limit(read);
           }
@@ -215,7 +213,7 @@ private int readInternal(byte[] b, int offset, int len) throws IOException {
     if (readAborted) {
       throw new IOException(readException);
     }
-    if (hasRemaining()) {
+    if (isEndOfStream()) {
       return -1;
     }
     len = Math.min(len, activeBuffer.remaining());