Merge branch 'master' of github.com:apache/spark into fix-concurrent-sql

Author: Andrew Or

common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java

@@ -194,8 +194,8 @@ private TransportClient createClient(InetSocketAddress address) throws IOExcepti
       .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, conf.connectionTimeoutMs())
       .option(ChannelOption.ALLOCATOR, pooledAllocator);
 
-    final AtomicReference<TransportClient> clientRef = new AtomicReference<TransportClient>();
-    final AtomicReference<Channel> channelRef = new AtomicReference<Channel>();
+    final AtomicReference<TransportClient> clientRef = new AtomicReference<>();
+    final AtomicReference<Channel> channelRef = new AtomicReference<>();
 
     bootstrap.handler(new ChannelInitializer<SocketChannel>() {
       @Override

common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java

@@ -64,8 +64,6 @@ public class ChunkFetchIntegrationSuite {
   static ManagedBuffer bufferChunk;
   static ManagedBuffer fileChunk;
 
-  private TransportConf transportConf;
-
   @BeforeClass
   public static void setUp() throws Exception {
     int bufSize = 100000;

common/network-common/src/test/java/org/apache/spark/network/RequestTimeoutIntegrationSuite.java

@@ -55,7 +55,7 @@ public class RequestTimeoutIntegrationSuite {
   private TransportConf conf;
 
   // A large timeout that "shouldn't happen", for the sake of faulty tests not hanging forever.
-  private final int FOREVER = 60 * 1000;
+  private static final int FOREVER = 60 * 1000;
 
   @Before
   public void setUp() throws Exception {

common/network-common/src/test/java/org/apache/spark/network/TransportClientFactorySuite.java

@@ -122,6 +122,8 @@ public void run() {
     for (TransportClient client : clients) {
       client.close();
     }
+
+    factory.close();
   }
 
   @Test

common/unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java

@@ -44,7 +44,8 @@
  * <p>
  * Note: This is not designed for general use cases, should not be used outside SQL.
  */
-public final class UTF8String implements Comparable<UTF8String>, Externalizable, KryoSerializable {
+public final class UTF8String implements Comparable<UTF8String>, Externalizable, KryoSerializable,
+  Cloneable {
 
   // These are only updated by readExternal() or read()
   @Nonnull

core/src/main/java/org/apache/spark/shuffle/sort/ShuffleExternalSorter.java

@@ -84,9 +84,9 @@ final class ShuffleExternalSorter extends MemoryConsumer {
    * this might not be necessary if we maintained a pool of re-usable pages in the TaskMemoryManager
    * itself).
    */
-  private final LinkedList<MemoryBlock> allocatedPages = new LinkedList<MemoryBlock>();
+  private final LinkedList<MemoryBlock> allocatedPages = new LinkedList<>();
 
-  private final LinkedList<SpillInfo> spills = new LinkedList<SpillInfo>();
+  private final LinkedList<SpillInfo> spills = new LinkedList<>();
 
   /** Peak memory used by this sorter so far, in bytes. **/
   private long peakMemoryUsedBytes;

core/src/main/java/org/apache/spark/status/api/v1/TaskSorting.java

@@ -29,7 +29,7 @@ public enum TaskSorting {
 
   private final Set<String> alternateNames;
   private TaskSorting(String... names) {
-    alternateNames = new HashSet<String>();
+    alternateNames = new HashSet<>();
     for (String n: names) {
       alternateNames.add(n);
     }

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeInMemorySorter.java

@@ -172,7 +172,7 @@ public void insertRecord(long recordPointer, long keyPrefix) {
     pos++;
   }
 
-  public final class SortedIterator extends UnsafeSorterIterator {
+  public final class SortedIterator extends UnsafeSorterIterator implements Cloneable {
 
     private final int numRecords;
     private int position;

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

@@ -376,8 +376,6 @@ private[spark] class MapOutputTrackerMaster(conf: SparkConf)
    * @param numReducers total number of reducers in the shuffle
    * @param fractionThreshold fraction of total map output size that a location must have
    *                          for it to be considered large.
-   *
-   * This method is not thread-safe.
    */
   def getLocationsWithLargestOutputs(
       shuffleId: Int,
@@ -386,28 +384,36 @@ private[spark] class MapOutputTrackerMaster(conf: SparkConf)
       fractionThreshold: Double)
     : Option[Array[BlockManagerId]] = {
 
-    if (mapStatuses.contains(shuffleId)) {
-      val statuses = mapStatuses(shuffleId)
-      if (statuses.nonEmpty) {
-        // HashMap to add up sizes of all blocks at the same location
-        val locs = new HashMap[BlockManagerId, Long]
-        var totalOutputSize = 0L
-        var mapIdx = 0
-        while (mapIdx < statuses.length) {
-          val status = statuses(mapIdx)
-          val blockSize = status.getSizeForBlock(reducerId)
-          if (blockSize > 0) {
-            locs(status.location) = locs.getOrElse(status.location, 0L) + blockSize
-            totalOutputSize += blockSize
+    val statuses = mapStatuses.get(shuffleId).orNull
+    if (statuses != null) {
+      statuses.synchronized {
+        if (statuses.nonEmpty) {
+          // HashMap to add up sizes of all blocks at the same location
+          val locs = new HashMap[BlockManagerId, Long]
+          var totalOutputSize = 0L
+          var mapIdx = 0
+          while (mapIdx < statuses.length) {
+            val status = statuses(mapIdx)
+            // status may be null here if we are called between registerShuffle, which creates an
+            // array with null entries for each output, and registerMapOutputs, which populates it
+            // with valid status entries. This is possible if one thread schedules a job which
+            // depends on an RDD which is currently being computed by another thread.
+            if (status != null) {
+              val blockSize = status.getSizeForBlock(reducerId)
+              if (blockSize > 0) {
+                locs(status.location) = locs.getOrElse(status.location, 0L) + blockSize
+                totalOutputSize += blockSize
+              }
+            }
+            mapIdx = mapIdx + 1
+          }
+          val topLocs = locs.filter { case (loc, size) =>
+            size.toDouble / totalOutputSize >= fractionThreshold
+          }
+          // Return if we have any locations which satisfy the required threshold
+          if (topLocs.nonEmpty) {
+            return Some(topLocs.keys.toArray)
           }
-          mapIdx = mapIdx + 1
-        }
-        val topLocs = locs.filter { case (loc, size) =>
-          size.toDouble / totalOutputSize >= fractionThreshold
-        }
-        // Return if we have any locations which satisfy the required threshold
-        if (topLocs.nonEmpty) {
-          return Some(topLocs.map(_._1).toArray)
         }
       }
     }

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -48,7 +48,7 @@ import org.apache.spark.util.{ChildFirstURLClassLoader, MutableURLClassLoader, U
 
 /**
  * Whether to submit, kill, or request the status of an application.
- * The latter two operations are currently supported only for standalone cluster mode.
+ * The latter two operations are currently supported only for standalone and Mesos cluster modes.
  */
 private[deploy] object SparkSubmitAction extends Enumeration {
   type SparkSubmitAction = Value