Refactored code and added test case.

In response to @srowen and @zsxwing's comments.

Author: tmagrino

core/src/main/scala/org/apache/spark/scheduler/TaskLocation.scala

@@ -64,18 +64,21 @@ private[spark] object TaskLocation {
 
   /**
    * Create a TaskLocation from a string returned by getPreferredLocations.
-   * These strings have the form [executorLocationTag][hostname][executorid], [hostname], or
+   * These strings have the form executor_[hostname]_[executorid], [hostname], or
    * hdfs_cache_[hostname], depending on whether the location is cached.
    */
   def apply(str: String): TaskLocation = {
     val hstr = str.stripPrefix(inMemoryLocationTag)
     if (hstr.equals(str)) {
       if (str.startsWith(executorLocationTag)) {
-        val splits = str.split("_", 3)
-        if (splits.length != 3) {
+        val hostAndExecutorId = str.stripPrefix(executorLocationTag)
+        val splits = hostAndExecutorId.split("_", 2)
+        if (splits.length != 2) {
           throw new IllegalArgumentException("Illegal executor location format: " + str)
         }
-        new ExecutorCacheTaskLocation(splits(1), splits(2))
+        val host = splits(0)
+        val executorId = splits(1)
+        new ExecutorCacheTaskLocation(host, executorId)
       } else {
         new HostTaskLocation(str)
       }

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

@@ -789,6 +789,8 @@ class TaskSetManagerSuite extends SparkFunSuite with LocalSparkContext with Logg
     assert(TaskLocation("host1") === HostTaskLocation("host1"))
     assert(TaskLocation("hdfs_cache_host1") === HDFSCacheTaskLocation("host1"))
     assert(TaskLocation("executor_host1_3") === ExecutorCacheTaskLocation("host1", "3"))
+    assert(TaskLocation("executor_some.host1_executor_task_3") ===
+      ExecutorCacheTaskLocation("some.host1", "executor_task_3"))
   }
 
   test("Kill other task attempts when one attempt belonging to the same task succeeds") {