Feedback.

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/security/CryptoStreamUtils.scala

@@ -67,12 +67,9 @@ private[spark] object CryptoStreamUtils extends Logging {
       key: Array[Byte]): WritableByteChannel = {
     val params = new CryptoParams(key, sparkConf)
     val iv = createInitializationVector(params.conf)
-    val buf = ByteBuffer.wrap(iv)
-    while (buf.hasRemaining()) {
-      channel.write(buf)
-    }
-
     val helper = new CryptoHelperChannel(channel)
+
+    helper.write(ByteBuffer.wrap(iv))
     new CryptoOutputStream(params.transformation, params.conf, helper, params.keySpec,
       new IvParameterSpec(iv))
   }

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -64,6 +64,11 @@ private[spark] trait BlockData {
 
   def toInputStream(): InputStream
 
+  /**
+   * Returns a Netty-friendly wrapper for the block's data.
+   *
+   * @see [[ManagedBuffer#convertToNetty()]]
+   */
   def toNetty(): Object
 
   def toChunkedByteBuffer(allocator: Int => ByteBuffer): ChunkedByteBuffer

core/src/main/scala/org/apache/spark/storage/BlockManagerManagedBuffer.scala

@@ -25,10 +25,13 @@ import org.apache.spark.network.buffer.ManagedBuffer
 import org.apache.spark.util.io.ChunkedByteBuffer
 
 /**
- * This [[ManagedBuffer]] wraps a ManagedBuffer retrieved from the [[BlockManager]]
+ * This [[ManagedBuffer]] wraps a [[BlockData]] instance retrieved from the [[BlockManager]]
  * so that the corresponding block's read lock can be released once this buffer's references
  * are released.
  *
+ * If `dispose` is set to try, the [[BlockData]]will be disposed when the buffer's reference
+ * count drops to zero.
+ *
  * This is effectively a wrapper / bridge to connect the BlockManager's notion of read locks
  * to the network layer's notion of retain / release counts.
  */