alteryx · markhamstra · Sep 12, 2016 · Jul 13, 2016 · Jul 12, 2016 · Jul 12, 2016
diff --git a/R/pkg/inst/tests/testthat/test_client.R b/R/pkg/inst/tests/testthat/test_client.R
@@ -32,7 +32,7 @@ test_that("no package specified doesn't add packages flag", {
 })
 
 test_that("multiple packages don't produce a warning", {
-  expect_that(generateSparkSubmitArgs("", "", "", "", c("A", "B")), not(gives_warning()))
+  expect_warning(generateSparkSubmitArgs("", "", "", "", c("A", "B")), NA)
 })
 
 test_that("sparkJars sparkPackages as character vectors", {

diff --git a/R/pkg/inst/tests/testthat/test_context.R b/R/pkg/inst/tests/testthat/test_context.R
@@ -109,6 +109,6 @@ test_that("sparkJars sparkPackages as comma-separated strings", {
 
   # check normalizePath
   f <- dir()[[1]]
-  expect_that(processSparkJars(f), not(gives_warning()))
+  expect_warning(processSparkJars(f), NA)
   expect_match(processSparkJars(f), f)
 })
diff --git a/R/pkg/inst/tests/testthat/test_sparkSQL.R b/R/pkg/inst/tests/testthat/test_sparkSQL.R
@@ -1119,9 +1119,9 @@ test_that("date functions on a DataFrame", {
                c(as.POSIXlt("2012-12-13 21:34:00 UTC"), as.POSIXlt("2014-12-15 10:24:34 UTC")))
   expect_equal(collect(select(df2, to_utc_timestamp(df2$b, "JST")))[, 1],
                c(as.POSIXlt("2012-12-13 03:34:00 UTC"), as.POSIXlt("2014-12-14 16:24:34 UTC")))
-  expect_more_than(collect(select(df2, unix_timestamp()))[1, 1], 0)
-  expect_more_than(collect(select(df2, unix_timestamp(df2$b)))[1, 1], 0)
-  expect_more_than(collect(select(df2, unix_timestamp(lit("2015-01-01"), "yyyy-MM-dd")))[1, 1], 0)
+  expect_gt(collect(select(df2, unix_timestamp()))[1, 1], 0)
+  expect_gt(collect(select(df2, unix_timestamp(df2$b)))[1, 1], 0)
+  expect_gt(collect(select(df2, unix_timestamp(lit("2015-01-01"), "yyyy-MM-dd")))[1, 1], 0)
 
   l3 <- list(list(a = 1000), list(a = -1000))
   df3 <- createDataFrame(sqlContext, l3)
@@ -1389,15 +1389,14 @@ test_that("toJSON() returns an RDD of the correct values", {
 
 test_that("showDF()", {
   df <- read.json(sqlContext, jsonPath)
-  s <- capture.output(showDF(df))
   expected <- paste("+----+-------+\n",
                     "| age|   name|\n",
                     "+----+-------+\n",
                     "|null|Michael|\n",
                     "|  30|   Andy|\n",
                     "|  19| Justin|\n",
                     "+----+-------+\n", sep = "")
-  expect_output(s, expected)
+  expect_output(showDF(df), expected)
 })
 
 test_that("isLocal()", {
@@ -1749,6 +1748,9 @@ test_that("Method as.data.frame as a synonym for collect()", {
   expect_equal(as.data.frame(irisDF), collect(irisDF))
   irisDF2 <- irisDF[irisDF$Species == "setosa", ]
   expect_equal(as.data.frame(irisDF2), collect(irisDF2))
+
+  # Make sure as.data.frame in the R base package is not covered
+  expect_error(as.data.frame(c(1, 2)), NA)
 })
 
 test_that("attach() on a DataFrame", {

diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationState.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationState.scala
@@ -22,6 +22,4 @@ private[master] object ApplicationState extends Enumeration {
   type ApplicationState = Value
 
   val WAITING, RUNNING, FINISHED, FAILED, KILLED, UNKNOWN = Value
-
-  val MAX_NUM_RETRY = 10
 }
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
@@ -71,6 +71,7 @@ private[deploy] class Master(
   private val RETAINED_DRIVERS = conf.getInt("spark.deploy.retainedDrivers", 200)
   private val REAPER_ITERATIONS = conf.getInt("spark.dead.worker.persistence", 15)
   private val RECOVERY_MODE = conf.get("spark.deploy.recoveryMode", "NONE")
+  private val MAX_EXECUTOR_RETRIES = conf.getInt("spark.deploy.maxExecutorRetries", 10)
 
   val workers = new HashSet[WorkerInfo]
   val idToApp = new HashMap[String, ApplicationInfo]
@@ -284,19 +285,20 @@ private[deploy] class Master(
 
             val normalExit = exitStatus == Some(0)
             // Only retry certain number of times so we don't go into an infinite loop.
-            if (!normalExit) {
-              if (appInfo.incrementRetryCount() < ApplicationState.MAX_NUM_RETRY) {
-                schedule()
-              } else {
-                val execs = appInfo.executors.values
-                if (!execs.exists(_.state == ExecutorState.RUNNING)) {
-                  logError(s"Application ${appInfo.desc.name} with ID ${appInfo.id} failed " +
-                    s"${appInfo.retryCount} times; removing it")
-                  removeApplication(appInfo, ApplicationState.FAILED)
-                }
+            // Important note: this code path is not exercised by tests, so be very careful when
+            // changing this `if` condition.
+            if (!normalExit
+                && appInfo.incrementRetryCount() >= MAX_EXECUTOR_RETRIES
+                && MAX_EXECUTOR_RETRIES >= 0) { // < 0 disables this application-killing path
+              val execs = appInfo.executors.values
+              if (!execs.exists(_.state == ExecutorState.RUNNING)) {
+                logError(s"Application ${appInfo.desc.name} with ID ${appInfo.id} failed " +
+                  s"${appInfo.retryCount} times; removing it")
+                removeApplication(appInfo, ApplicationState.FAILED)
               }
             }
           }
+          schedule()
         }
         case None =>
           logWarning(s"Got status update for unknown executor $appId/$execId")

diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala b/core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.rpc.netty
 
-import java.util.concurrent.{ThreadPoolExecutor, ConcurrentHashMap, LinkedBlockingQueue, TimeUnit}
+import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap, LinkedBlockingQueue, ThreadPoolExecutor, TimeUnit}
 import javax.annotation.concurrent.GuardedBy
 
 import scala.collection.JavaConverters._
@@ -41,8 +41,10 @@ private[netty] class Dispatcher(nettyEnv: NettyRpcEnv) extends Logging {
     val inbox = new Inbox(ref, endpoint)
   }
 
-  private val endpoints = new ConcurrentHashMap[String, EndpointData]
-  private val endpointRefs = new ConcurrentHashMap[RpcEndpoint, RpcEndpointRef]
+  private val endpoints: ConcurrentMap[String, EndpointData] =
+    new ConcurrentHashMap[String, EndpointData]
+  private val endpointRefs: ConcurrentMap[RpcEndpoint, RpcEndpointRef] =
+    new ConcurrentHashMap[RpcEndpoint, RpcEndpointRef]
 
   // Track the receivers whose inboxes may contain messages.
   private val receivers = new LinkedBlockingQueue[EndpointData]

diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -364,14 +364,15 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
     conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))
   }
 
-  // Called by subclasses when notified of a lost worker
-  def removeExecutor(executorId: String, reason: ExecutorLossReason) {
-    try {
-      driverEndpoint.askWithRetry[Boolean](RemoveExecutor(executorId, reason))
-    } catch {
-      case e: Exception =>
-        throw new SparkException("Error notifying standalone scheduler's driver endpoint", e)
-    }
+  /**
+   * Called by subclasses when notified of a lost worker. It just fires the message and returns
+   * at once.
+   */
+  protected def removeExecutor(executorId: String, reason: ExecutorLossReason): Unit = {
+    // Only log the failure since we don't care about the result.
+    driverEndpoint.ask[Boolean](RemoveExecutor(executorId, reason)).onFailure { case t =>
+      logError(t.getMessage, t)
+    }(ThreadUtils.sameThread)
   }
 
   def sufficientResourcesRegistered(): Boolean = true

diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/AllJobsResource.scala b/core/src/main/scala/org/apache/spark/status/api/v1/AllJobsResource.scala
@@ -86,7 +86,7 @@ private[v1] object AllJobsResource {
         numTasks = job.numTasks,
         numActiveTasks = job.numActiveTasks,
         numCompletedTasks = job.numCompletedTasks,
-        numSkippedTasks = job.numCompletedTasks,
+        numSkippedTasks = job.numSkippedTasks,
         numFailedTasks = job.numFailedTasks,
         numActiveStages = job.numActiveStages,
         numCompletedStages = job.completedStageIndices.size,

diff --git a/core/src/main/scala/org/apache/spark/ui/env/EnvironmentPage.scala b/core/src/main/scala/org/apache/spark/ui/env/EnvironmentPage.scala
@@ -26,11 +26,15 @@ import org.apache.spark.ui.{UIUtils, WebUIPage}
 private[ui] class EnvironmentPage(parent: EnvironmentTab) extends WebUIPage("") {
   private val listener = parent.listener
 
+  private def removePass(kv: (String, String)): (String, String) = {
+    if (kv._1.toLowerCase.contains("password")) (kv._1, "******") else kv
+  }
+
   def render(request: HttpServletRequest): Seq[Node] = {
     val runtimeInformationTable = UIUtils.listingTable(
       propertyHeader, jvmRow, listener.jvmInformation, fixedWidth = true)
     val sparkPropertiesTable = UIUtils.listingTable(
-      propertyHeader, propertyRow, listener.sparkProperties, fixedWidth = true)
+      propertyHeader, propertyRow, listener.sparkProperties.map(removePass), fixedWidth = true)
     val systemPropertiesTable = UIUtils.listingTable(
       propertyHeader, propertyRow, listener.systemProperties, fixedWidth = true)
     val classpathEntriesTable = UIUtils.listingTable(

diff --git a/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala b/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
@@ -592,7 +592,9 @@ private[spark] class ExternalSorter[K, V, C](
       val ds = deserializeStream
       deserializeStream = null
       fileStream = null
-      ds.close()
+      if (ds != null) {
+        ds.close()
+      }
       // NOTE: We don't do file.delete() here because that is done in ExternalSorter.stop().
       // This should also be fixed in ExternalAppendOnlyMap.
     }

diff --git a/...ources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_1__expectation.json b/...ources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_1__expectation.json
@@ -6,7 +6,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,

diff --git a/...ources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_2__expectation.json b/...ources/HistoryServerExpectations/job_list_from_multi_attempt_app_json_2__expectation.json
@@ -6,7 +6,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,

diff --git a/core/src/test/resources/HistoryServerExpectations/job_list_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/job_list_json_expectation.json
@@ -6,7 +6,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,
@@ -20,7 +20,7 @@
   "numTasks" : 16,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 15,
-  "numSkippedTasks" : 15,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 1,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,
@@ -34,7 +34,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,

diff --git a/core/src/test/resources/HistoryServerExpectations/one_job_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/one_job_json_expectation.json
@@ -6,7 +6,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,

diff --git a/.../test/resources/HistoryServerExpectations/succeeded_failed_job_list_json_expectation.json b/.../test/resources/HistoryServerExpectations/succeeded_failed_job_list_json_expectation.json
@@ -6,7 +6,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,
@@ -20,7 +20,7 @@
   "numTasks" : 16,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 15,
-  "numSkippedTasks" : 15,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 1,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,
@@ -34,7 +34,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,

diff --git a/core/src/test/resources/HistoryServerExpectations/succeeded_job_list_json_expectation.json b/core/src/test/resources/HistoryServerExpectations/succeeded_job_list_json_expectation.json
@@ -6,7 +6,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,
@@ -20,7 +20,7 @@
   "numTasks" : 8,
   "numActiveTasks" : 0,
   "numCompletedTasks" : 8,
-  "numSkippedTasks" : 8,
+  "numSkippedTasks" : 0,
   "numFailedTasks" : 0,
   "numActiveStages" : 0,
   "numCompletedStages" : 1,

diff --git a/core/src/test/scala/org/apache/spark/DistributedSuite.scala b/core/src/test/scala/org/apache/spark/DistributedSuite.scala
@@ -131,6 +131,21 @@ class DistributedSuite extends SparkFunSuite with Matchers with LocalSparkContex
     }
   }
 
+  test("repeatedly failing task that crashes JVM with a zero exit code (SPARK-16925)") {
+    // Ensures that if a task which causes the JVM to exit with a zero exit code will cause the
+    // Spark job to eventually fail.
+    sc = new SparkContext(clusterUrl, "test")
+    failAfter(Span(100000, Millis)) {
+      val thrown = intercept[SparkException] {
+        sc.parallelize(1 to 1, 1).foreachPartition { _ => System.exit(0) }
+      }
+      assert(thrown.getClass === classOf[SparkException])
+      assert(thrown.getMessage.contains("failed 4 times"))
+    }
+    // Check that the cluster is still usable:
+    sc.parallelize(1 to 10).count()
+  }
+
   test("caching") {
     sc = new SparkContext(clusterUrl, "test")
     val data = sc.parallelize(1 to 1000, 10).cache()

diff --git a/dev/create-release/release-build.sh b/dev/create-release/release-build.sh
@@ -80,7 +80,7 @@ NEXUS_PROFILE=d63f592e7eac0 # Profile for Spark staging uploads
 BASE_DIR=$(pwd)
 
 MVN="build/mvn --force"
-PUBLISH_PROFILES="-Pyarn -Phive -Phadoop-2.2"
+PUBLISH_PROFILES="-Pyarn -Phive -Phive-thriftserver -Phadoop-2.2"
 PUBLISH_PROFILES="$PUBLISH_PROFILES -Pspark-ganglia-lgpl -Pkinesis-asl"
 
 rm -rf spark
@@ -187,7 +187,7 @@ if [[ "$1" == "package" ]]; then
   # We increment the Zinc port each time to avoid OOM's and other craziness if multiple builds
   # share the same Zinc server.
   make_binary_release "hadoop1" "-Psparkr -Phadoop-1 -Phive -Phive-thriftserver" "3030" &
-  make_binary_release "hadoop1-scala2.11" "-Psparkr -Phadoop-1 -Phive -Dscala-2.11" "3031" &
+  make_binary_release "hadoop1-scala2.11" "-Psparkr -Phadoop-1 -Phive -Phive-thriftserver -Dscala-2.11" "3031" &
   make_binary_release "cdh4" "-Psparkr -Phadoop-1 -Phive -Phive-thriftserver -Dhadoop.version=2.0.0-mr1-cdh4.2.0" "3032" &
   make_binary_release "hadoop2.3" "-Psparkr -Phadoop-2.3 -Phive -Phive-thriftserver -Pyarn" "3033" &
   make_binary_release "hadoop2.4" "-Psparkr -Phadoop-2.4 -Phive -Phive-thriftserver -Pyarn" "3034" &
@@ -256,8 +256,7 @@ if [[ "$1" == "publish-snapshot" ]]; then
   # Generate random point for Zinc
   export ZINC_PORT=$(python -S -c "import random; print random.randrange(3030,4030)")
 
-  $MVN -DzincPort=$ZINC_PORT --settings $tmp_settings -DskipTests $PUBLISH_PROFILES \
-    -Phive-thriftserver deploy
+  $MVN -DzincPort=$ZINC_PORT --settings $tmp_settings -DskipTests $PUBLISH_PROFILES deploy
   ./dev/change-scala-version.sh 2.11
   $MVN -DzincPort=$ZINC_PORT -Dscala-2.11 --settings $tmp_settings \
     -DskipTests $PUBLISH_PROFILES clean deploy
@@ -293,8 +292,7 @@ if [[ "$1" == "publish-release" ]]; then
   # Generate random point for Zinc
   export ZINC_PORT=$(python -S -c "import random; print random.randrange(3030,4030)")
 
-  $MVN -DzincPort=$ZINC_PORT -Dmaven.repo.local=$tmp_repo -DskipTests $PUBLISH_PROFILES \
-    -Phive-thriftserver clean install
+  $MVN -DzincPort=$ZINC_PORT -Dmaven.repo.local=$tmp_repo -DskipTests $PUBLISH_PROFILES clean install
 
   ./dev/change-scala-version.sh 2.11
 

diff --git a/dev/deps/spark-deps-hadoop-1 b/dev/deps/spark-deps-hadoop-1
@@ -53,7 +53,7 @@ curator-recipes-2.4.0.jar
 datanucleus-api-jdo-3.2.6.jar
 datanucleus-core-3.2.10.jar
 datanucleus-rdbms-3.2.9.jar
-derby-10.10.1.1.jar
+derby-10.12.1.1.jar
 eigenbase-properties-1.1.5.jar
 geronimo-annotation_1.0_spec-1.1.1.jar
 geronimo-jaspic_1.0_spec-1.0.jar
@@ -143,7 +143,7 @@ servlet-api-2.5.jar
 slf4j-api-1.7.10.jar
 slf4j-log4j12-1.7.10.jar
 snappy-0.2.jar
-snappy-java-1.1.2.1.jar
+snappy-java-1.1.2.6.jar
 spire-macros_2.10-0.7.4.jar
 spire_2.10-0.7.4.jar
 stax-api-1.0.1.jar

diff --git a/dev/deps/spark-deps-hadoop-2.2 b/dev/deps/spark-deps-hadoop-2.2
@@ -53,7 +53,7 @@ curator-recipes-2.4.0.jar
 datanucleus-api-jdo-3.2.6.jar
 datanucleus-core-3.2.10.jar
 datanucleus-rdbms-3.2.9.jar
-derby-10.10.1.1.jar
+derby-10.12.1.1.jar
 eigenbase-properties-1.1.5.jar
 geronimo-annotation_1.0_spec-1.1.1.jar
 geronimo-jaspic_1.0_spec-1.0.jar
@@ -173,7 +173,7 @@ servlet-api-2.5.jar
 slf4j-api-1.7.10.jar
 slf4j-log4j12-1.7.10.jar
 snappy-0.2.jar
-snappy-java-1.1.2.1.jar
+snappy-java-1.1.2.6.jar
 spire-macros_2.10-0.7.4.jar
 spire_2.10-0.7.4.jar
 stax-api-1.0.1.jar

diff --git a/dev/deps/spark-deps-hadoop-2.3 b/dev/deps/spark-deps-hadoop-2.3
@@ -54,7 +54,7 @@ curator-recipes-2.4.0.jar
 datanucleus-api-jdo-3.2.6.jar
 datanucleus-core-3.2.10.jar
 datanucleus-rdbms-3.2.9.jar
-derby-10.10.1.1.jar
+derby-10.12.1.1.jar
 eigenbase-properties-1.1.5.jar
 geronimo-annotation_1.0_spec-1.1.1.jar
 geronimo-jaspic_1.0_spec-1.0.jar
@@ -164,7 +164,7 @@ servlet-api-2.5.jar
 slf4j-api-1.7.10.jar
 slf4j-log4j12-1.7.10.jar
 snappy-0.2.jar
-snappy-java-1.1.2.1.jar
+snappy-java-1.1.2.6.jar
 spire-macros_2.10-0.7.4.jar
 spire_2.10-0.7.4.jar
 stax-api-1.0-2.jar