Shivaram's code review comment.

Author: rxin

core/src/main/scala/org/apache/spark/network/netty/server/BlockHeaderEncoder.scala

@@ -29,19 +29,19 @@ class BlockHeaderEncoder extends MessageToByteEncoder[BlockHeader] {
   override def encode(ctx: ChannelHandlerContext, msg: BlockHeader, out: ByteBuf): Unit = {
     // message = message length (4 bytes) + block id length (4 bytes) + block id + block data
     // message length = block id length (4 bytes) + size of block id + size of block data
-    val blockId = msg.blockId.getBytes
+    val blockIdBytes = msg.blockId.getBytes
     msg.error match {
       case Some(errorMsg) =>
         val errorBytes = errorMsg.getBytes
-        out.writeInt(4 + blockId.length + errorBytes.size)
-        out.writeInt(-blockId.length)
-        out.writeBytes(blockId)
-        out.writeBytes(errorBytes)
+        out.writeInt(4 + blockIdBytes.length + errorBytes.size)
+        out.writeInt(-blockIdBytes.length)  // use negative block id length to represent errors
+        out.writeBytes(blockIdBytes)  // next is blockId itself
+        out.writeBytes(errorBytes)  // error message
       case None =>
-        val blockId = msg.blockId.getBytes
-        out.writeInt(4 + blockId.length + msg.blockSize)
-        out.writeInt(blockId.length)
-        out.writeBytes(blockId)
+        out.writeInt(4 + blockIdBytes.length + msg.blockSize)
+        out.writeInt(blockIdBytes.length)  // First 4 bytes is blockId length
+        out.writeBytes(blockIdBytes)  // next is blockId itself
+        // msg of size blockSize will be written by ServerHandler
     }
   }
 }

core/src/main/scala/org/apache/spark/network/netty/server/BlockServerChannelInitializer.scala

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.netty.server
+
+import io.netty.channel.ChannelInitializer
+import io.netty.channel.socket.SocketChannel
+import io.netty.handler.codec.LineBasedFrameDecoder
+import io.netty.handler.codec.string.StringDecoder
+import io.netty.util.CharsetUtil
+import org.apache.spark.storage.BlockDataProvider
+
+
+/** Channel initializer that sets up the pipeline for the BlockServer. */
+private[netty]
+class BlockServerChannelInitializer(dataProvider: BlockDataProvider)
+  extends ChannelInitializer[SocketChannel] {
+
+  override def initChannel(ch: SocketChannel): Unit = {
+    ch.pipeline
+      .addLast("frameDecoder", new LineBasedFrameDecoder(1024))  // max block id length 1024
+      .addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8))
+      .addLast("blockHeaderEncoder", new BlockHeaderEncoder)
+      .addLast("handler", new BlockServerHandler(dataProvider))
+  }
+}

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

@@ -277,6 +277,8 @@ object BlockFetcherIterator {
 
       bytesInFlight += req.size
       val sizeMap = req.blocks.toMap // so we can look up the size of each blockID
+
+      // This could throw a TimeoutException. In that case we will just retry the task.
       val client = blockManager.nettyBlockClientFactory.createClient(
         cmId.host, req.address.nettyPort)
       val blocks = req.blocks.map(_._1.toString)
@@ -304,8 +306,10 @@ object BlockFetcherIterator {
             }
           }))
 
-          readMetrics.remoteBytesRead += blockSize
-          readMetrics.remoteBlocksFetched += 1
+          readMetrics.synchronized {
+            readMetrics.remoteBytesRead += blockSize
+            readMetrics.remoteBlocksFetched += 1
+          }
           logDebug("Got remote block " + blockId + " after " + Utils.getUsedTimeMs(startTime))
         },
         (blockId: String, errorMsg: String) => {