Shorten NioManagedBuffer and NettyManagedBuffer class names.

Author: rxin

core/src/main/scala/org/apache/spark/network/ManagedBuffer.scala

@@ -36,11 +36,11 @@ import org.apache.spark.util.{ByteBufferInputStream, Utils}
  * should specify how the data is provided:
  *
  * - [[FileSegmentManagedBuffer]]: data backed by part of a file
- * - [[NioByteBufferManagedBuffer]]: data backed by a NIO ByteBuffer
- * - [[NettyByteBufManagedBuffer]]: data backed by a Netty ByteBuf
+ * - [[NioManagedBuffer]]: data backed by a NIO ByteBuffer
+ * - [[NettyManagedBuffer]]: data backed by a Netty ByteBuf
  *
  * The concrete buffer implementation might be managed outside the JVM garbage collector.
- * For example, in the case of [[NettyByteBufManagedBuffer]], the buffers are reference counted.
+ * For example, in the case of [[NettyManagedBuffer]], the buffers are reference counted.
  * In that case, if the buffer is going to be passed around to a different thread, retain/release
  * should be called.
  */
@@ -149,7 +149,7 @@ final class FileSegmentManagedBuffer(val file: File, val offset: Long, val lengt
 /**
  * A [[ManagedBuffer]] backed by [[java.nio.ByteBuffer]].
  */
-final class NioByteBufferManagedBuffer(buf: ByteBuffer) extends ManagedBuffer {
+final class NioManagedBuffer(buf: ByteBuffer) extends ManagedBuffer {
 
   override def size: Long = buf.remaining()
 
@@ -168,7 +168,7 @@ final class NioByteBufferManagedBuffer(buf: ByteBuffer) extends ManagedBuffer {
 /**
  * A [[ManagedBuffer]] backed by a Netty [[ByteBuf]].
  */
-final class NettyByteBufManagedBuffer(buf: ByteBuf) extends ManagedBuffer {
+final class NettyManagedBuffer(buf: ByteBuf) extends ManagedBuffer {
 
   override def size: Long = buf.readableBytes()
 

core/src/main/scala/org/apache/spark/network/netty/protocol.scala

@@ -25,7 +25,7 @@ import io.netty.channel.ChannelHandler.Sharable
 import io.netty.handler.codec._
 
 import org.apache.spark.Logging
-import org.apache.spark.network.{NettyByteBufManagedBuffer, ManagedBuffer}
+import org.apache.spark.network.{NettyManagedBuffer, ManagedBuffer}
 
 
 /** Messages from the client to the server. */
@@ -141,7 +141,7 @@ final class ClientRequestDecoder extends MessageToMessageDecoder[ByteBuf] {
       case 1 =>  // BlockUploadRequest
         val blockId = ProtocolUtils.readBlockId(in)
         in.retain()  // retain the bytebuf so we don't recycle it immediately.
-        BlockUploadRequest(blockId, new NettyByteBufManagedBuffer(in))
+        BlockUploadRequest(blockId, new NettyManagedBuffer(in))
     }
 
     assert(decoded.id == msgTypeId)
@@ -218,7 +218,7 @@ final class ServerResponseDecoder extends MessageToMessageDecoder[ByteBuf] {
       case 0 =>  // BlockFetchSuccess
         val blockId = ProtocolUtils.readBlockId(in)
         in.retain()
-        new BlockFetchSuccess(blockId, new NettyByteBufManagedBuffer(in))
+        new BlockFetchSuccess(blockId, new NettyManagedBuffer(in))
 
       case 1 =>  // BlockFetchFailure
         val blockId = ProtocolUtils.readBlockId(in)

core/src/main/scala/org/apache/spark/network/nio/NioBlockTransferService.scala

@@ -104,7 +104,7 @@ final class NioBlockTransferService(conf: SparkConf, securityManager: SecurityMa
           val blockId = blockMessage.getId
           val networkSize = blockMessage.getData.limit()
           listener.onBlockFetchSuccess(
-            blockId.toString, new NioByteBufferManagedBuffer(blockMessage.getData))
+            blockId.toString, new NioManagedBuffer(blockMessage.getData))
         }
       }
     }(cm.futureExecContext)
@@ -189,7 +189,7 @@ final class NioBlockTransferService(conf: SparkConf, securityManager: SecurityMa
   private def putBlock(blockId: String, bytes: ByteBuffer, level: StorageLevel) {
     val startTimeMs = System.currentTimeMillis()
     logDebug("PutBlock " + blockId + " started from " + startTimeMs + " with data: " + bytes)
-    blockDataManager.putBlockData(blockId, new NioByteBufferManagedBuffer(bytes), level)
+    blockDataManager.putBlockData(blockId, new NioManagedBuffer(bytes), level)
     logDebug("PutBlock " + blockId + " used " + Utils.getUsedTimeMs(startTimeMs)
       + " with data size: " + bytes.limit)
   }

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

@@ -223,7 +223,7 @@ private[spark] class BlockManager(
       val blockBytesOpt = doGetLocal(bid, asBlockResult = false).asInstanceOf[Option[ByteBuffer]]
       if (blockBytesOpt.isDefined) {
         val buffer = blockBytesOpt.get
-        new NioByteBufferManagedBuffer(buffer)
+        new NioManagedBuffer(buffer)
       } else {
         throw new BlockNotFoundException(blockId)
       }
@@ -868,7 +868,7 @@ private[spark] class BlockManager(
             data.rewind()
             logTrace(s"Trying to replicate $blockId of ${data.limit()} bytes to $peer")
             blockTransferService.uploadBlockSync(
-              peer.host, peer.port, blockId.toString, new NioByteBufferManagedBuffer(data), tLevel)
+              peer.host, peer.port, blockId.toString, new NioManagedBuffer(data), tLevel)
             logTrace(s"Replicated $blockId of ${data.limit()} bytes to $peer in %f ms"
               .format((System.currentTimeMillis - onePeerStartTime)))
             peersReplicatedTo += peer

core/src/test/scala/org/apache/spark/network/netty/BlockClientHandlerSuite.scala

@@ -64,7 +64,7 @@ class BlockClientHandlerSuite extends FunSuite with PrivateMethodTester {
     buf.put(blockData.getBytes)
     buf.flip()
 
-    channel.writeInbound(BlockFetchSuccess(blockId, new NioByteBufferManagedBuffer(buf)))
+    channel.writeInbound(BlockFetchSuccess(blockId, new NioManagedBuffer(buf)))
 
     assert(parsedBlockId === blockId)
     assert(parsedBlockData === blockData)
@@ -119,7 +119,7 @@ class BlockClientHandlerSuite extends FunSuite with PrivateMethodTester {
     assert(sizeOfOutstandingRequests(handler) === 3)
 
     val channel = new EmbeddedChannel(handler)
-    channel.writeInbound(BlockFetchSuccess("b1", new NettyByteBufManagedBuffer(Unpooled.buffer())))
+    channel.writeInbound(BlockFetchSuccess("b1", new NettyManagedBuffer(Unpooled.buffer())))
     // Need to figure out a way to generate an exception
     assert(successCount.get() === 1)
     assert(errorCount.get() === 2)

core/src/test/scala/org/apache/spark/network/netty/ServerClientIntegrationSuite.scala

@@ -65,7 +65,7 @@ class ServerClientIntegrationSuite extends FunSuite with BeforeAndAfterAll {
     server = new BlockServer(new NettyConfig(new SparkConf), new BlockDataManager {
       override def getBlockData(blockId: String): ManagedBuffer = {
         if (blockId == bufferBlockId) {
-          new NioByteBufferManagedBuffer(buf)
+          new NioManagedBuffer(buf)
         } else if (blockId == fileBlockId) {
           new FileSegmentManagedBuffer(testFile, 10, testFile.length - 25)
         } else {

core/src/test/scala/org/apache/spark/network/netty/TestManagedBuffer.scala

@@ -22,7 +22,7 @@ import java.nio.ByteBuffer
 
 import io.netty.buffer.Unpooled
 
-import org.apache.spark.network.{NettyByteBufManagedBuffer, ManagedBuffer}
+import org.apache.spark.network.{NettyManagedBuffer, ManagedBuffer}
 
 
 /**
@@ -36,7 +36,7 @@ class TestManagedBuffer(len: Int) extends ManagedBuffer {
 
   private val byteArray: Array[Byte] = Array.tabulate[Byte](len)(_.toByte)
 
-  private val underlying = new NettyByteBufManagedBuffer(Unpooled.wrappedBuffer(byteArray))
+  private val underlying = new NettyManagedBuffer(Unpooled.wrappedBuffer(byteArray))
 
   override def size: Long = underlying.size