Revert "update upstream (apache-spark-on-k8s#366)" (apache-spark-on-k8s#367)

This reverts commit a3c2539.

Author: pwoody

common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java

@@ -32,7 +32,6 @@
 import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
-import org.apache.commons.lang3.SystemUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -99,8 +98,7 @@ private void init(String hostToBind, int portToBind) {
       .group(bossGroup, workerGroup)
       .channel(NettyUtils.getServerChannelClass(ioMode))
       .option(ChannelOption.ALLOCATOR, allocator)
-      .childOption(ChannelOption.ALLOCATOR, allocator)
-      .childOption(ChannelOption.SO_REUSEADDR, !SystemUtils.IS_OS_WINDOWS);
+      .childOption(ChannelOption.ALLOCATOR, allocator);
 
     this.metrics = new NettyMemoryMetrics(
       allocator, conf.getModuleName() + "-server", conf);

common/unsafe/src/main/java/org/apache/spark/unsafe/array/ByteArrayMethods.java

@@ -33,11 +33,7 @@ public static long nextPowerOf2(long num) {
   }
 
   public static int roundNumberOfBytesToNearestWord(int numBytes) {
-    return (int)roundNumberOfBytesToNearestWord((long)numBytes);
-  }
-
-  public static long roundNumberOfBytesToNearestWord(long numBytes) {
-    long remainder = numBytes & 0x07;  // This is equivalent to `numBytes % 8`
+    int remainder = numBytes & 0x07;  // This is equivalent to `numBytes % 8`
     if (remainder == 0) {
       return numBytes;
     } else {

common/unsafe/src/test/java/org/apache/spark/unsafe/memory/MemoryBlockSuite.java

@@ -120,8 +120,6 @@ private void check(MemoryBlock memory, Object obj, long offset, int length) {
     } catch (Exception expected) {
       Assert.assertThat(expected.getMessage(), containsString("should not be larger than"));
     }
-
-    memory.setPageNumber(MemoryBlock.NO_PAGE_NUMBER);
   }
 
   @Test
@@ -167,13 +165,11 @@ public void testOffHeapArrayMemoryBlock() {
     int length = 56;
 
     check(memory, obj, offset, length);
-    memoryAllocator.free(memory);
 
     long address = Platform.allocateMemory(112);
     memory = new OffHeapMemoryBlock(address, length);
     obj = memory.getBaseObject();
     offset = memory.getBaseOffset();
     check(memory, obj, offset, length);
-    Platform.freeMemory(address);
   }
 }

core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala

@@ -108,7 +108,6 @@ private[spark] abstract class RestSubmissionServer(
         resolvedConnectionFactories: _*)
     connector.setHost(host)
     connector.setPort(startPort)
-    connector.setReuseAddress(!Utils.isWindows)
     server.addConnector(connector)
 
     val mainHandler = new ServletContextHandler

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

@@ -287,7 +287,7 @@ private[spark] class TaskSetManager(
     None
   }
 
-  /** Check whether a task once ran an attempt on a given host */
+  /** Check whether a task is currently running an attempt on a given host */
   private def hasAttemptOnHost(taskIndex: Int, host: String): Boolean = {
     taskAttempts(taskIndex).exists(_.host == host)
   }

core/src/main/scala/org/apache/spark/ui/JettyUtils.scala

@@ -344,7 +344,6 @@ private[spark] object JettyUtils extends Logging {
           connectionFactories: _*)
         connector.setPort(port)
         connector.setHost(hostName)
-        connector.setReuseAddress(!Utils.isWindows)
 
         // Currently we only use "SelectChannelConnector"
         // Limit the max acceptor number to 8 so that we don't waste a lot of threads

core/src/test/scala/org/apache/spark/SparkContextSuite.scala

@@ -20,7 +20,7 @@ package org.apache.spark
 import java.io.File
 import java.net.{MalformedURLException, URI}
 import java.nio.charset.StandardCharsets
-import java.util.concurrent.{CountDownLatch, Semaphore, TimeUnit}
+import java.util.concurrent.{Semaphore, TimeUnit}
 
 import scala.concurrent.duration._
 
@@ -498,36 +498,45 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
 
   test("Cancelling stages/jobs with custom reasons.") {
     sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
-    sc.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "true")
     val REASON = "You shall not pass"
+    val slices = 10
 
-    for (cancelWhat <- Seq("stage", "job")) {
-      // This countdown latch used to make sure stage or job canceled in listener
-      val latch = new CountDownLatch(1)
-
-      val listener = cancelWhat match {
-        case "stage" =>
-          new SparkListener {
-            override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
-              sc.cancelStage(taskStart.stageId, REASON)
-              latch.countDown()
-            }
+    val listener = new SparkListener {
+      override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
+        if (SparkContextSuite.cancelStage) {
+          eventually(timeout(10.seconds)) {
+            assert(SparkContextSuite.isTaskStarted)
           }
-        case "job" =>
-          new SparkListener {
-            override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
-              sc.cancelJob(jobStart.jobId, REASON)
-              latch.countDown()
-            }
+          sc.cancelStage(taskStart.stageId, REASON)
+          SparkContextSuite.cancelStage = false
+          SparkContextSuite.semaphore.release(slices)
+        }
+      }
+
+      override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
+        if (SparkContextSuite.cancelJob) {
+          eventually(timeout(10.seconds)) {
+            assert(SparkContextSuite.isTaskStarted)
           }
+          sc.cancelJob(jobStart.jobId, REASON)
+          SparkContextSuite.cancelJob = false
+          SparkContextSuite.semaphore.release(slices)
+        }
       }
-      sc.addSparkListener(listener)
+    }
+    sc.addSparkListener(listener)
+
+    for (cancelWhat <- Seq("stage", "job")) {
+      SparkContextSuite.semaphore.drainPermits()
+      SparkContextSuite.isTaskStarted = false
+      SparkContextSuite.cancelStage = (cancelWhat == "stage")
+      SparkContextSuite.cancelJob = (cancelWhat == "job")
 
       val ex = intercept[SparkException] {
-        sc.range(0, 10000L, numSlices = 10).mapPartitions { x =>
-          x.synchronized {
-            x.wait()
-          }
+        sc.range(0, 10000L, numSlices = slices).mapPartitions { x =>
+          SparkContextSuite.isTaskStarted = true
+          // Block waiting for the listener to cancel the stage or job.
+          SparkContextSuite.semaphore.acquire()
           x
         }.count()
       }
@@ -541,11 +550,9 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
           fail("Expected the cause to be SparkException, got " + cause.toString() + " instead.")
       }
 
-      latch.await(20, TimeUnit.SECONDS)
       eventually(timeout(20.seconds)) {
         assert(sc.statusTracker.getExecutorInfos.map(_.numRunningTasks()).sum == 0)
       }
-      sc.removeSparkListener(listener)
     }
   }
 
@@ -630,6 +637,8 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
 }
 
 object SparkContextSuite {
+  @volatile var cancelJob = false
+  @volatile var cancelStage = false
   @volatile var isTaskStarted = false
   @volatile var taskKilled = false
   @volatile var taskSucceeded = false

external/kafka-0-8/src/main/scala/org/apache/spark/streaming/kafka/KafkaRDD.scala

@@ -191,7 +191,6 @@ class KafkaRDD[
 
     private def fetchBatch: Iterator[MessageAndOffset] = {
       val req = new FetchRequestBuilder()
-        .clientId(consumer.clientId)
         .addFetch(part.topic, part.partition, requestOffset, kc.config.fetchMessageMaxBytes)
         .build()
       val resp = consumer.fetch(req)

mllib/src/main/resources/META-INF/services/org.apache.spark.ml.util.MLFormatRegister

@@ -1,4 +1,2 @@
 org.apache.spark.ml.regression.InternalLinearRegressionModelWriter
-org.apache.spark.ml.regression.PMMLLinearRegressionModelWriter
-org.apache.spark.ml.clustering.InternalKMeansModelWriter
-org.apache.spark.ml.clustering.PMMLKMeansModelWriter
+org.apache.spark.ml.regression.PMMLLinearRegressionModelWriter

mllib/src/main/scala/org/apache/spark/ml/clustering/KMeans.scala

@@ -17,13 +17,11 @@
 
 package org.apache.spark.ml.clustering
 
-import scala.collection.mutable
-
 import org.apache.hadoop.fs.Path
 
 import org.apache.spark.SparkException
 import org.apache.spark.annotation.{Experimental, Since}
-import org.apache.spark.ml.{Estimator, Model, PipelineStage}
+import org.apache.spark.ml.{Estimator, Model}
 import org.apache.spark.ml.linalg.{Vector, VectorUDT}
 import org.apache.spark.ml.param._
 import org.apache.spark.ml.param.shared._
@@ -32,7 +30,7 @@ import org.apache.spark.mllib.clustering.{DistanceMeasure, KMeans => MLlibKMeans
 import org.apache.spark.mllib.linalg.{Vector => OldVector, Vectors => OldVectors}
 import org.apache.spark.mllib.linalg.VectorImplicits._
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
+import org.apache.spark.sql.{DataFrame, Dataset, Row}
 import org.apache.spark.sql.functions.{col, udf}
 import org.apache.spark.sql.types.{IntegerType, StructType}
 import org.apache.spark.storage.StorageLevel
@@ -105,8 +103,8 @@ private[clustering] trait KMeansParams extends Params with HasMaxIter with HasFe
 @Since("1.5.0")
 class KMeansModel private[ml] (
     @Since("1.5.0") override val uid: String,
-    private[clustering] val parentModel: MLlibKMeansModel)
-  extends Model[KMeansModel] with KMeansParams with GeneralMLWritable {
+    private val parentModel: MLlibKMeansModel)
+  extends Model[KMeansModel] with KMeansParams with MLWritable {
 
   @Since("1.5.0")
   override def copy(extra: ParamMap): KMeansModel = {
@@ -154,14 +152,14 @@ class KMeansModel private[ml] (
   }
 
   /**
-   * Returns a [[org.apache.spark.ml.util.GeneralMLWriter]] instance for this ML instance.
+   * Returns a [[org.apache.spark.ml.util.MLWriter]] instance for this ML instance.
    *
    * For [[KMeansModel]], this does NOT currently save the training [[summary]].
    * An option to save [[summary]] may be added in the future.
    *
    */
   @Since("1.6.0")
-  override def write: GeneralMLWriter = new GeneralMLWriter(this)
+  override def write: MLWriter = new KMeansModel.KMeansModelWriter(this)
 
   private var trainingSummary: Option[KMeansSummary] = None
 
@@ -187,47 +185,6 @@ class KMeansModel private[ml] (
   }
 }
 
-/** Helper class for storing model data */
-private case class ClusterData(clusterIdx: Int, clusterCenter: Vector)
-
-
-/** A writer for KMeans that handles the "internal" (or default) format */
-private class InternalKMeansModelWriter extends MLWriterFormat with MLFormatRegister {
-
-  override def format(): String = "internal"
-  override def stageName(): String = "org.apache.spark.ml.clustering.KMeansModel"
-
-  override def write(path: String, sparkSession: SparkSession,
-    optionMap: mutable.Map[String, String], stage: PipelineStage): Unit = {
-    val instance = stage.asInstanceOf[KMeansModel]
-    val sc = sparkSession.sparkContext
-    // Save metadata and Params
-    DefaultParamsWriter.saveMetadata(instance, path, sc)
-    // Save model data: cluster centers
-    val data: Array[ClusterData] = instance.clusterCenters.zipWithIndex.map {
-      case (center, idx) =>
-        ClusterData(idx, center)
-    }
-    val dataPath = new Path(path, "data").toString
-    sparkSession.createDataFrame(data).repartition(1).write.parquet(dataPath)
-  }
-}
-
-/** A writer for KMeans that handles the "pmml" format */
-private class PMMLKMeansModelWriter extends MLWriterFormat with MLFormatRegister {
-
-  override def format(): String = "pmml"
-  override def stageName(): String = "org.apache.spark.ml.clustering.KMeansModel"
-
-  override def write(path: String, sparkSession: SparkSession,
-    optionMap: mutable.Map[String, String], stage: PipelineStage): Unit = {
-    val instance = stage.asInstanceOf[KMeansModel]
-    val sc = sparkSession.sparkContext
-    instance.parentModel.toPMML(sc, path)
-  }
-}
-
-
 @Since("1.6.0")
 object KMeansModel extends MLReadable[KMeansModel] {
 
@@ -237,12 +194,30 @@ object KMeansModel extends MLReadable[KMeansModel] {
   @Since("1.6.0")
   override def load(path: String): KMeansModel = super.load(path)
 
+  /** Helper class for storing model data */
+  private case class Data(clusterIdx: Int, clusterCenter: Vector)
+
   /**
    * We store all cluster centers in a single row and use this class to store model data by
    * Spark 1.6 and earlier. A model can be loaded from such older data for backward compatibility.
    */
   private case class OldData(clusterCenters: Array[OldVector])
 
+  /** [[MLWriter]] instance for [[KMeansModel]] */
+  private[KMeansModel] class KMeansModelWriter(instance: KMeansModel) extends MLWriter {
+
+    override protected def saveImpl(path: String): Unit = {
+      // Save metadata and Params
+      DefaultParamsWriter.saveMetadata(instance, path, sc)
+      // Save model data: cluster centers
+      val data: Array[Data] = instance.clusterCenters.zipWithIndex.map { case (center, idx) =>
+        Data(idx, center)
+      }
+      val dataPath = new Path(path, "data").toString
+      sparkSession.createDataFrame(data).repartition(1).write.parquet(dataPath)
+    }
+  }
+
   private class KMeansModelReader extends MLReader[KMeansModel] {
 
     /** Checked against metadata when loading model */
@@ -257,7 +232,7 @@ object KMeansModel extends MLReadable[KMeansModel] {
       val dataPath = new Path(path, "data").toString
 
       val clusterCenters = if (majorVersion(metadata.sparkVersion) >= 2) {
-        val data: Dataset[ClusterData] = sparkSession.read.parquet(dataPath).as[ClusterData]
+        val data: Dataset[Data] = sparkSession.read.parquet(dataPath).as[Data]
         data.collect().sortBy(_.clusterIdx).map(_.clusterCenter).map(OldVectors.fromML)
       } else {
         // Loads KMeansModel stored with the old format used by Spark 1.6 and earlier.