Fix no ending loop

core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala

@@ -70,7 +70,7 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
   def writeWithSlice(channel: WritableByteChannel): Unit = {
     for (bytes <- getChunks()) {
       val capacity = bytes.limit()
-      while (bytes.position() < capacity) {
+      while (bytes.position() < capacity && bytes.remaining() > 0) {
         val ioSize = Math.min(bytes.remaining(), NIO_BUFFER_LIMIT.toInt)
         bytes.limit(bytes.position + ioSize)
         channel.write(bytes)

core/src/test/scala/org/apache/spark/io/ChunkedByteBufferSuite.scala

@@ -56,6 +56,23 @@ class ChunkedByteBufferSuite extends SparkFunSuite {
     assert(chunkedByteBuffer.getChunks().head.position() === 0)
   }
 
+  test("benchmark") {
+    val buffer100 = ByteBuffer.allocate(1024 * 1024 * 100)
+    val buffer30 = ByteBuffer.allocate(1024 * 1024 * 30)
+    val chunkedByteBuffer = new ChunkedByteBuffer(Array.fill(5)(buffer100))
+    var starTime = System.currentTimeMillis()
+    for (i <- 1 to 10) {
+      chunkedByteBuffer.writeFully(new ByteArrayWritableChannel(chunkedByteBuffer.size.toInt))
+    }
+    // scalastyle:off
+    System.out.println(System.currentTimeMillis() - starTime)
+    starTime = System.currentTimeMillis()
+    for (i <- 1 to 10) {
+      chunkedByteBuffer.writeWithSlice(new ByteArrayWritableChannel(chunkedByteBuffer.size.toInt))
+    }
+    System.out.println(System.currentTimeMillis() - starTime)
+  }
+
   test("toArray()") {
     val empty = ByteBuffer.wrap(Array.empty[Byte])
     val bytes = ByteBuffer.wrap(Array.tabulate(8)(_.toByte))