HDFS-3246: Addressing PR comments

Author: Sahil Takiar

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/CryptoInputStream.java

@@ -329,19 +329,18 @@ public void close() throws IOException {
   public int read(long position, byte[] buffer, int offset, int length)
       throws IOException {
     checkStream();
-    if (in instanceof PositionedReadable) {
-      final int n = ((PositionedReadable) in).read(position, buffer, offset, 
-          length);
-      if (n > 0) {
-        // This operation does not change the current offset of the file
-        decrypt(position, buffer, offset, n);
-      }
-      
-      return n;
-    } else {
+    if (!(in instanceof PositionedReadable)) {
       throw new UnsupportedOperationException("This stream does not support " +
           "positioned read.");
     }
+    final int n = ((PositionedReadable) in).read(position, buffer, offset,
+        length);
+    if (n > 0) {
+      // This operation does not change the current offset of the file
+      decrypt(position, buffer, offset, n);
+    }
+
+    return n;
   }
 
   /**
@@ -351,19 +350,18 @@ public int read(long position, byte[] buffer, int offset, int length)
   public int read(long position, final ByteBuffer buf)
       throws IOException {
     checkStream();
-    if (in instanceof ByteBufferPositionedReadable) {
-      int bufPos = buf.position();
-      final int n = ((ByteBufferPositionedReadable) in).read(position, buf);
-      if (n > 0) {
-        // This operation does not change the current offset of the file
-        decrypt(position, buf, n, bufPos);
-      }
-
-      return n;
-    } else {
+    if (!(in instanceof ByteBufferPositionedReadable)) {
       throw new UnsupportedOperationException("This stream does not support " +
           "positioned reads with byte buffers.");
     }
+    int bufPos = buf.position();
+    final int n = ((ByteBufferPositionedReadable) in).read(position, buf);
+    if (n > 0) {
+      // This operation does not change the current offset of the file
+      decrypt(position, buf, n, bufPos);
+    }
+
+    return n;
   }
 
   /**
@@ -481,16 +479,15 @@ private void decrypt(long filePosition, ByteBuffer buf, int length, int start)
   public void readFully(long position, byte[] buffer, int offset, int length)
       throws IOException {
     checkStream();
-    if (in instanceof PositionedReadable) {
-      ((PositionedReadable) in).readFully(position, buffer, offset, length);
-      if (length > 0) {
-        // This operation does not change the current offset of the file
-        decrypt(position, buffer, offset, length);
-      }
-    } else {
+    if (!(in instanceof PositionedReadable)) {
       throw new UnsupportedOperationException("This stream does not support " +
           "positioned readFully.");
     }
+    ((PositionedReadable) in).readFully(position, buffer, offset, length);
+    if (length > 0) {
+      // This operation does not change the current offset of the file
+      decrypt(position, buffer, offset, length);
+    }
   }
 
   @Override
@@ -515,13 +512,12 @@ public void seek(long pos) throws IOException {
         outBuffer.position(outBuffer.position() + forward);
       }
     } else {
-      if (in instanceof Seekable) {
-        ((Seekable) in).seek(pos);
-        resetStreamOffset(pos);
-      } else {
+      if (!(in instanceof Seekable)) {
         throw new UnsupportedOperationException("This stream does not " +
                 "support seek.");
       }
+      ((Seekable) in).seek(pos);
+      resetStreamOffset(pos);
     }
   }
 
@@ -675,14 +671,13 @@ public boolean seekToNewSource(long targetPos) throws IOException {
     Preconditions.checkArgument(targetPos >= 0, 
         "Cannot seek to negative offset.");
     checkStream();
-    if (in instanceof Seekable) {
-      boolean result = ((Seekable) in).seekToNewSource(targetPos);
-      resetStreamOffset(targetPos);
-      return result;
-    } else {
+    if (!(in instanceof Seekable)) {
       throw new UnsupportedOperationException("This stream does not support " +
           "seekToNewSource.");
     }
+    boolean result = ((Seekable) in).seekToNewSource(targetPos);
+    resetStreamOffset(targetPos);
+    return result;
   }
 
   @Override
@@ -691,63 +686,58 @@ public ByteBuffer read(ByteBufferPool bufferPool, int maxLength,
       UnsupportedOperationException {
     checkStream();
     if (outBuffer.remaining() > 0) {
-      if (in instanceof Seekable) {
-        // Have some decrypted data unread, need to reset.
-        ((Seekable) in).seek(getPos());
-        resetStreamOffset(getPos());
-      } else {
+      if (!(in instanceof Seekable)) {
         throw new UnsupportedOperationException("This stream does not " +
                 "support seek.");
       }
+      // Have some decrypted data unread, need to reset.
+      ((Seekable) in).seek(getPos());
+      resetStreamOffset(getPos());
     }
-    if (in instanceof HasEnhancedByteBufferAccess) {
-      final ByteBuffer buffer = ((HasEnhancedByteBufferAccess) in).
-          read(bufferPool, maxLength, opts);
-      if (buffer != null) {
-        final int n = buffer.remaining();
-        if (n > 0) {
-          streamOffset += buffer.remaining(); // Read n bytes
-          final int pos = buffer.position();
-          decrypt(buffer, n, pos);
-        }
-      }
-      return buffer;
-    } else {
+    if (!(in instanceof HasEnhancedByteBufferAccess)) {
       throw new UnsupportedOperationException("This stream does not support " +
           "enhanced byte buffer access.");
     }
+    final ByteBuffer buffer = ((HasEnhancedByteBufferAccess) in).
+        read(bufferPool, maxLength, opts);
+    if (buffer != null) {
+      final int n = buffer.remaining();
+      if (n > 0) {
+        streamOffset += buffer.remaining(); // Read n bytes
+        final int pos = buffer.position();
+        decrypt(buffer, n, pos);
+      }
+    }
+    return buffer;
   }
 
   @Override
   public void releaseBuffer(ByteBuffer buffer) {
-    if (in instanceof HasEnhancedByteBufferAccess) {
-      ((HasEnhancedByteBufferAccess) in).releaseBuffer(buffer);
-    } else {
+    if (!(in instanceof HasEnhancedByteBufferAccess)) {
       throw new UnsupportedOperationException("This stream does not support " + 
           "release buffer.");
     }
+    ((HasEnhancedByteBufferAccess) in).releaseBuffer(buffer);
   }
 
   @Override
   public void setReadahead(Long readahead) throws IOException,
       UnsupportedOperationException {
-    if (in instanceof CanSetReadahead) {
-      ((CanSetReadahead) in).setReadahead(readahead);
-    } else {
+    if (!(in instanceof CanSetReadahead)) {
       throw new UnsupportedOperationException("This stream does not support " +
           "setting the readahead caching strategy.");
     }
+    ((CanSetReadahead) in).setReadahead(readahead);
   }
 
   @Override
   public void setDropBehind(Boolean dropCache) throws IOException,
       UnsupportedOperationException {
-    if (in instanceof CanSetReadahead) {
-      ((CanSetDropBehind) in).setDropBehind(dropCache);
-    } else {
+    if (!(in instanceof CanSetReadahead)) {
       throw new UnsupportedOperationException("This stream does not " +
           "support setting the drop-behind caching setting.");
     }
+    ((CanSetDropBehind) in).setDropBehind(dropCache);
   }
 
   @Override
@@ -851,12 +841,11 @@ public boolean hasCapability(String capability) {
     case StreamCapabilities.DROPBEHIND:
     case StreamCapabilities.READBYTEBUFFER:
     case StreamCapabilities.PREADBYTEBUFFER:
-      if (in instanceof StreamCapabilities) {
-        return ((StreamCapabilities) in).hasCapability(capability);
-      } else {
+      if (!(in instanceof StreamCapabilities)) {
         throw new UnsupportedOperationException("This stream does not expose " +
           "its stream capabilities.");
       }
+      return ((StreamCapabilities) in).hasCapability(capability);
     default:
       return false;
     }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ByteBufferPositionedReadable.java

@@ -41,10 +41,10 @@ public interface ByteBufferPositionedReadable {
    * should be written to.
    * <p>
    * After a successful call, {@code buf.position()} will be advanced by the
-   * number of bytes read and {@code buf.limit()} should be unchanged.
+   * number of bytes read and {@code buf.limit()} will be unchanged.
    * <p>
    * In the case of an exception, the state of the buffer (the contents of the
-   * buffer, the {@code buf.position()}, the {@code buf.limit(), etc.) is
+   * buffer, the {@code buf.position()}, the {@code buf.limit()}, etc.) is
    * undefined, and callers should be prepared to recover from this
    * eventuality.
    * <p>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ByteBufferReadable.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 
@@ -33,11 +34,11 @@ public interface ByteBufferReadable {
    * Reads up to buf.remaining() bytes into buf. Callers should use
    * buf.limit(..) to control the size of the desired read.
    * <p>
-   * After a successful call, buf.position() will be advanced by the number 
-   * of bytes read and buf.limit() should be unchanged.
+   * After a successful call, {@code buf.position()} will be advanced by the
+   * number of bytes read and {@code buf.limit()} will be unchanged.
    * <p>
    * In the case of an exception, the state of the buffer (the contents of the
-   * buffer, the {@code buf.position()}, the {@code buf.limit(), etc.) is
+   * buffer, the {@code buf.position()}, the {@code buf.limit()}, etc.) is
    * undefined, and callers should be prepared to recover from this
    * eventuality.
    * <p>

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs-tests/test_libhdfs_ops.c

@@ -260,6 +260,17 @@ int main(int argc, char **argv) {
         // read path
         hdfsFileDisableDirectRead(readFile);
 
+        if (hdfsFileUsesDirectRead(readFile)) {
+            fprintf(stderr, "Disabled direct reads, but it is still enabled");
+            shutdown_and_exit(cl, -1);
+        }
+
+        if (!hdfsFileUsesDirectPread(readFile)) {
+            fprintf(stderr, "Disabled direct reads, but direct preads was "
+                            "disabled as well");
+            shutdown_and_exit(cl, -1);
+        }
+
         num_read_bytes = hdfsRead(fs, readFile, (void*)buffer,
                 sizeof(buffer));
         if (strncmp(fileContents, buffer, strlen(fileContents)) != 0) {
@@ -353,6 +364,17 @@ int main(int argc, char **argv) {
         // read path
         hdfsFileDisableDirectPread(preadFile);
 
+        if (hdfsFileUsesDirectPread(preadFile)) {
+            fprintf(stderr, "Disabled direct preads, but it is still enabled");
+            shutdown_and_exit(cl, -1);
+        }
+
+        if (!hdfsFileUsesDirectRead(preadFile)) {
+            fprintf(stderr, "Disabled direct preads, but direct read was "
+                            "disabled as well");
+            shutdown_and_exit(cl, -1);
+        }
+
         num_pread_bytes = hdfsPread(fs, preadFile, 0, (void*)buffer, sizeof(buffer));
         if (strncmp(fileContents, buffer, strlen(fileContents)) != 0) {
             fprintf(stderr, "Failed to pread. Expected %s but got %s (%d bytes)\n",

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/hdfs.c

@@ -40,7 +40,7 @@
 
 // Bit fields for hdfsFile_internal flags
 #define HDFS_FILE_SUPPORTS_DIRECT_READ (1<<0)
-#define HDFS_FILE_SUPPORTS_DIRECT_PREAD (1<<0)
+#define HDFS_FILE_SUPPORTS_DIRECT_PREAD (1<<1)
 
 /**
  * Reads bytes using the read(ByteBuffer) API. By using Java
@@ -1011,6 +1011,19 @@ int hdfsStreamBuilderSetDefaultBlockSize(struct hdfsStreamBuilder *bld,
     return 0;
 }
 
+/**
+ * Delegates to FsDataInputStream#hasCapability(String). Used to check if a
+ * given input stream supports certain methods, such as
+ * ByteBufferReadable#read(ByteBuffer).
+ *
+ * @param jFile the FsDataInputStream to call hasCapability on
+ * @param capability the name of the capability to query; for a full list of
+ *        possible values see StreamCapabilities
+ *
+ * @return true if the given jFile has the given capability, false otherwise
+ *
+ * @see org.apache.hadoop.fs.StreamCapabilities
+ */
 static int hdfsHasStreamCapability(jobject jFile,
         const char *capability) {
     int ret = 0;
@@ -1033,7 +1046,7 @@ static int hdfsHasStreamCapability(jobject jFile,
     }
     jthr = invokeMethod(env, &jVal, INSTANCE, jFile,
             JC_FS_DATA_INPUT_STREAM, "hasCapability", "(Ljava/lang/String;)Z",
-	    jCapabilityString);
+            jCapabilityString);
     if (jthr) {
         ret = printExceptionAndFree(env, jthr, PRINT_EXC_ALL,
                 "hdfsHasStreamCapability(%s): FSDataInputStream#hasCapability",
@@ -1568,7 +1581,16 @@ tSize readDirect(hdfsFS fs, hdfsFile f, void* buffer, tSize length)
             "readDirect: FSDataInputStream#read");
         return -1;
     }
-    return (jVal.i < 0) ? 0 : jVal.i;
+    // Reached EOF, return 0
+    if (jVal.i < 0) {
+        return 0;
+    }
+    // 0 bytes read, return error
+    if (jVal.i == 0) {
+        errno = EINTR;
+        return -1;
+    }
+    return jVal.i;
 }
 
 /**
@@ -1693,7 +1715,16 @@ tSize preadDirect(hdfsFS fs, hdfsFile f, tOffset position, void* buffer,
            "preadDirect: FSDataInputStream#read");
        return -1;
     }
-    return (jVal.i < 0) ? 0 : jVal.i;
+    // Reached EOF, return 0
+    if (jVal.i < 0) {
+        return 0;
+    }
+    // 0 bytes read, return error
+    if (jVal.i == 0) {
+        errno = EINTR;
+        return -1;
+    }
+    return jVal.i;
 }
 
 tSize hdfsWrite(hdfsFS fs, hdfsFile f, const void* buffer, tSize length)