83ac9a4bbf
Current version of StreamingContext.stop() directly kills all the data receivers (NetworkReceiver) without waiting for the data already received to be persisted and processed. This PR provides the fix. Now, when the StreamingContext.stop() is called, the following sequence of steps will happen. 1. The driver will send a stop signal to all the active receivers. 2. Each receiver, when it gets a stop signal from the driver, first stop receiving more data, then waits for the thread that persists data blocks to BlockManager to finish persisting all receive data, and finally quits. 3. After all the receivers have stopped, the driver will wait for the Job Generator and Job Scheduler to finish processing all the received data. It also fixes the semantics of StreamingContext.start and stop. It will throw appropriate errors and warnings if stop() is called before start(), stop() is called twice, etc. Author: Tathagata Das <tathagata.das1565@gmail.com> Closes #247 from tdas/graceful-shutdown and squashes the following commits: 61c0016 [Tathagata Das] Updated MIMA binary check excludes. ae1d39b [Tathagata Das] Merge remote-tracking branch 'apache-github/master' into graceful-shutdown 6b59cfc [Tathagata Das] Minor changes based on Andrew's comment on PR. d0b8d65 [Tathagata Das] Reduced time taken by graceful shutdown unit test. f55bc67 [Tathagata Das] Fix scalastyle c69b3a7 [Tathagata Das] Updates based on Patrick's comments. c43b8ae [Tathagata Das] Added graceful shutdown to Spark Streaming.
86 lines
3.4 KiB
Scala
86 lines
3.4 KiB
Scala
/*
|
|
* Licensed to the Apache Software Foundation (ASF) under one or more
|
|
* contributor license agreements. See the NOTICE file distributed with
|
|
* this work for additional information regarding copyright ownership.
|
|
* The ASF licenses this file to You under the Apache License, Version 2.0
|
|
* (the "License"); you may not use this file except in compliance with
|
|
* the License. You may obtain a copy of the License at
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
* See the License for the specific language governing permissions and
|
|
* limitations under the License.
|
|
*/
|
|
|
|
import com.typesafe.tools.mima.plugin.MimaKeys.{binaryIssueFilters, previousArtifact}
|
|
import com.typesafe.tools.mima.plugin.MimaPlugin.mimaDefaultSettings
|
|
import sbt._
|
|
|
|
object MimaBuild {
|
|
|
|
def ignoredABIProblems(base: File) = {
|
|
import com.typesafe.tools.mima.core._
|
|
import com.typesafe.tools.mima.core.ProblemFilters._
|
|
|
|
// Excludes placed here will be used for all Spark versions
|
|
val defaultExcludes = Seq()
|
|
|
|
// Read package-private excludes from file
|
|
val excludeFilePath = (base.getAbsolutePath + "/.mima-excludes")
|
|
val excludeFile = file(excludeFilePath)
|
|
val packagePrivateList: Seq[String] =
|
|
if (!excludeFile.exists()) {
|
|
Seq()
|
|
} else {
|
|
IO.read(excludeFile).split("\n")
|
|
}
|
|
|
|
def excludeClass(className: String) = {
|
|
Seq(
|
|
excludePackage(className),
|
|
ProblemFilters.exclude[MissingClassProblem](className),
|
|
ProblemFilters.exclude[MissingTypesProblem](className),
|
|
excludePackage(className + "$"),
|
|
ProblemFilters.exclude[MissingClassProblem](className + "$"),
|
|
ProblemFilters.exclude[MissingTypesProblem](className + "$")
|
|
)
|
|
}
|
|
def excludeSparkClass(className: String) = excludeClass("org.apache.spark." + className)
|
|
|
|
val packagePrivateExcludes = packagePrivateList.flatMap(excludeClass)
|
|
|
|
/* Excludes specific to a given version of Spark. When comparing the given version against
|
|
its immediate predecessor, the excludes listed here will be applied. */
|
|
val versionExcludes =
|
|
SparkBuild.SPARK_VERSION match {
|
|
case v if v.startsWith("1.0") =>
|
|
Seq(
|
|
excludePackage("org.apache.spark.api.java"),
|
|
excludePackage("org.apache.spark.streaming.api.java"),
|
|
excludePackage("org.apache.spark.mllib")
|
|
) ++
|
|
excludeSparkClass("rdd.ClassTags") ++
|
|
excludeSparkClass("util.XORShiftRandom") ++
|
|
excludeSparkClass("mllib.recommendation.MFDataGenerator") ++
|
|
excludeSparkClass("mllib.optimization.SquaredGradient") ++
|
|
excludeSparkClass("mllib.regression.RidgeRegressionWithSGD") ++
|
|
excludeSparkClass("mllib.regression.LassoWithSGD") ++
|
|
excludeSparkClass("mllib.regression.LinearRegressionWithSGD") ++
|
|
excludeSparkClass("streaming.dstream.NetworkReceiver") ++
|
|
excludeSparkClass("streaming.dstream.NetworkReceiver#NetworkReceiverActor")
|
|
case _ => Seq()
|
|
}
|
|
|
|
defaultExcludes ++ packagePrivateExcludes ++ versionExcludes
|
|
}
|
|
|
|
def mimaSettings(sparkHome: File) = mimaDefaultSettings ++ Seq(
|
|
previousArtifact := None,
|
|
binaryIssueFilters ++= ignoredABIProblems(sparkHome)
|
|
)
|
|
|
|
}
|