review feedback

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

@@ -59,7 +59,7 @@ private[spark] trait ExecutorAllocationClient {
    * @param adjustTargetNumExecutors whether the target number of executors will be adjusted down
    *                                 after these executors have been killed
    * @param countFailures if there are tasks running on the executors when they are killed, whether
-   *                      those failures be counted to task failure limits?
+    *                     to count those failures toward task failure limits
    * @param force whether to force kill busy executors, default false
    * @return the ids of the executors acknowledged by the cluster manager to be removed.
    */

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

@@ -337,10 +337,10 @@ private[spark] class ExecutorAllocationManager(
       // If the new target has not changed, avoid sending a message to the cluster manager
       if (numExecutorsTarget < oldNumExecutorsTarget) {
         // We lower the target number of executors but don't actively kill any yet.  Killing is
-        // controlled separately by an idle timeout.  Its still helpful to reduce the target number
+        // controlled separately by an idle timeout.  It's still helpful to reduce the target number
         // in case an executor just happens to get lost (eg., bad hardware, or the cluster manager
         // preempts it) -- in that case, there is no point in trying to immediately  get a new
-        // executor, since we couldn't even use it yet.
+        // executor, since we wouldn't even use it yet.
         client.requestTotalExecutors(numExecutorsTarget, localityAwareTasks, hostToLocalTaskCount)
         logDebug(s"Lowering target number of executors to $numExecutorsTarget (previously " +
           s"$oldNumExecutorsTarget) because not all requested executors are actually needed")

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

@@ -1687,7 +1687,7 @@ class SparkContext(config: SparkConf) extends Logging {
   private[spark] def killAndReplaceExecutor(executorId: String): Boolean = {
     schedulerBackend match {
       case b: ExecutorAllocationClient =>
-        b.killExecutors(Seq(executorId), adjustTargetNumExecutors = false, countFailures = false,
+        b.killExecutors(Seq(executorId), adjustTargetNumExecutors = false, countFailures = true,
           force = true).nonEmpty
       case _ =>
         logWarning("Killing executors is not supported by current scheduler.")