| Property Name | Default | Meaning |
|---|---|---|
spark.hadoop.yarn.timeline.domain |
+ + | + If UI permissions are set through `spark.acls.enable` or `spark.ui.acls.enable` being true, + the optional name of a predefined timeline domain to use . If unset, + a value is created programmatically. + | +
spark.hadoop.yarn.timeline.post.retry.interval |
+ 1s | ++ Interval in milliseconds between POST retries. Every + failure adds another delay of this interval before the next retry + attempt. That is, first 1s, then 2s, 3s, ... + | +
spark.hadoop.yarn.timeline.batch.size |
+ 3 | ++ How many events to batch up before submitting them to the timeline service. + This is a performance optimization. + | +
spark.hadoop.yarn.timeline.post.limit |
+ 1000 | ++ Limit on number of queued events to posts. When exceeded + new events will be dropped. This is to place a limit on how much + memory will be consumed if the timeline server goes offline. + | +
spark.hadoop.yarn.timeline.shutdown.waittime |
+ 30s | ++ Maximum time in to wait for event posting to complete when the service stops. + | +
spark.hadoop.yarn.timeline.listen |
+ true | ++ This flag exists for testing: if `false` the history publishing + service will not register for events with the spark context. As + a result, lifecycle events will not be picked up. + | +
| Property Name | Default | Meaning |
|---|---|---|
spark.history.yarn.window.limit |
+ 24h | ++ The earliest time to look for events. The default value is 24 hours. + This property limits the amount of data queried off the YARN timeline server; + applications started before this window will not be checked to see if they have completed. + Set this to 0 for no limits (and increased load on the timeline server). + | +
spark.history.yarn.backround.refresh.interval |
+ 60s | ++ The interval between background refreshes of the history data. + A value of 0s means "no background updates: manual refreshes only". + | +
spark.history.yarn.manual.refresh.interval |
+ 30s | ++ Minimum interval between manual refreshes of the history data; refreshing the + page before this limit will not trigger an update. + A value of 0s means "page refreshes do not trigger updates of the application list" + | +
spark.history.yarn.event-fetch-limit |
+ 1000 | ++ Maximum number of application histories to fetch + from the timeline server in a single GET request. + | +
spark.history.yarn.diagnostics |
+ false | ++ A flag to indicate whether low-level diagnostics information should be included in + status pages. This is for debugging and diagnostics. + | +
spark.history.yarn.probe.running.applications |
+ true | ++ Should the history provider query the YARN Resource Manager to verify that + incompleted applications are actually still running. + | +
entityType have
+ * default values; None for optional, empty
+ * collections for the others.
+ * @param entityType entity type
+ * @param primaryFilter primary filter
+ * @param secondaryFilters map of secondary filters
+ * @param fields list of fields to retrieve
+ * @param limit limit on how many to retrieve
+ * @param windowStart time window to start retrieval
+ * @param windowEnd time window to stop retrieval
+ * @param fromId optional ID to start from
+ * @param fromTs optional timestamp to start from
+ * @return a possibly empty list of entities
+ */
+ def listEntities(
+ entityType: String,
+ primaryFilter: Option[(String, String)] = None,
+ secondaryFilters: Map[String, String] = Map(),
+ fields: Seq[String] = Nil,
+ limit: Option[Long] = None,
+ windowStart: Option[Long] = None,
+ windowEnd: Option[Long] = None,
+ fromId: Option[String] = None,
+ fromTs: Option[Long] = None): List[TimelineEntity] = {
+ require(!entityType.isEmpty, "no entity type")
+ var resource = entityResource(entityType)
+ // build the resource
+ // every application returns a new result, which complicates applying map and list arguments
+ // to it. hence the use of a variable
+ resource = primaryFilter match {
+ case Some((key, value)) =>
+ resource.queryParam("primaryFilter", s"$key:$value")
+ case None =>
+ resource
+ }
+ secondaryFilters foreach
+ ((t: (String, String)) =>
+ resource = resource.queryParam("secondaryFilter", s"${t._1}:${t._2}"))
+ resource = applyOptionalParam(resource, "windowStart", windowStart)
+ resource = applyOptionalParam(resource, "windowEnd", windowEnd)
+ resource = applyOptionalParam(resource, "limit", limit)
+ resource = applyOptionalParam(resource, "fromId", fromId)
+ resource = applyOptionalParam(resource, "fromTs", fromTs)
+ if (fields.nonEmpty) {
+ resource = resource.queryParam("fields", fields.mkString(","))
+ }
+ // execute the request
+ val response = get(resource.getURI,
+ () => resource.accept(APPLICATION_JSON).get(classOf[TimelineEntities]))
+ response.getEntities.asScala.toList
+ }
+
+ /**
+ * Get the resource of an entity type
+ * @param entityType entity type
+ * @return resource
+ */
+ def entityResource(entityType: String): WebResource = {
+ subresource(entityType)
+ }
+
+ /**
+ * Get an entity
+ * @param entityType type
+ * @param entityId the entity
+ * @return the entity if it was found
+ */
+ def getEntity(entityType: String, entityId: String): TimelineEntity = {
+ require(!entityId.isEmpty, "no entity ID")
+ val resource = entityResource(entityType).path(entityId)
+ get(resource.getURI,
+ () => resource.accept(APPLICATION_JSON).get(classOf[TimelineEntity]))
+ }
+
+ /**
+ * toString method returns the URI of the timeline service
+ * @return
+ */
+ override def toString: String = {
+ s"Timeline Query Client against $timelineURI"
+ }
+
+ /**
+ * Get the time the token was last renewed
+ * @return a system timestamp of the last renewal; 0 on startup
+ */
+ def lastTokenRenewal: Long = {
+ _lastTokenRenewal.get()
+ }
+
+ /**
+ * Get count of token renewals
+ * @return
+ */
+ def tokenRenewalCount: Long = {
+ _tokenRenewalCount.get()
+ }
+}
+
+/**
+ * Simple About response. The timeline V1 API keeps this type hidden in the server code, even though
+ * it is tagged `@Public`
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+private[spark] class AboutResponse {
+
+ var other: Map[String, Object] = Map()
+
+ var About: String = _
+
+ @JsonAnySetter
+ def handleUnknown(key: String, value: Object): Unit = {
+ other += (key -> value)
+ }
+
+}
+
+/**
+ * Constants associated with the Query API
+ */
+private[spark] object TimelineQueryClient {
+ val MESSAGE_CHECK_URL = "Check the URL of the timeline service:"
+ val MESSAGE_EMPTY_RESPONSE = s"No data in the response"
+ val EVENTS = "EVENTS"
+ val LAST_EVENT_ONLY = "LASTEVENTONLY"
+ val OTHER_INFO = "OTHERINFO"
+ val PRIMARY_FILTERS = "PRIMARYFILTERS"
+ val RELATED_ENTITIES = "RELATEDENTITIES"
+}
diff --git a/yarn/src/history/main/scala/org/apache/spark/deploy/history/yarn/server/YarnHistoryProvider.scala b/yarn/src/history/main/scala/org/apache/spark/deploy/history/yarn/server/YarnHistoryProvider.scala
new file mode 100644
index 000000000000..4bb939f7f630
--- /dev/null
+++ b/yarn/src/history/main/scala/org/apache/spark/deploy/history/yarn/server/YarnHistoryProvider.scala
@@ -0,0 +1,1389 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.server
+
+import java.io.{FileNotFoundException, IOException, InterruptedIOException}
+import java.net.URI
+import java.util.Date
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
+import java.util.concurrent.{LinkedBlockingQueue, TimeUnit}
+import java.util.zip.ZipOutputStream
+
+import scala.collection.JavaConverters._
+
+import com.codahale.metrics.{Counter, Gauge, Metric, MetricRegistry, Timer}
+import org.apache.hadoop.security.UserGroupInformation
+import org.apache.hadoop.service.{Service, ServiceOperations}
+import org.apache.hadoop.yarn.api.records.ApplicationReport
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity
+import org.apache.hadoop.yarn.client.api.YarnClient
+import org.apache.hadoop.yarn.conf.YarnConfiguration
+
+import org.apache.spark.{Logging, SecurityManager, SparkConf, SparkException}
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.deploy.history.{ApplicationHistoryProvider, HistoryServer}
+import org.apache.spark.deploy.history.yarn.YarnHistoryService._
+import org.apache.spark.deploy.history.yarn.{ExtendedMetricsSource, YarnTimelineUtils}
+import org.apache.spark.deploy.history.yarn.YarnTimelineUtils._
+import org.apache.spark.deploy.history.yarn.rest.JerseyBinding
+import org.apache.spark.deploy.history.yarn.server.TimelineQueryClient._
+import org.apache.spark.deploy.history.yarn.server.YarnProviderUtils._
+import org.apache.spark.scheduler.{ApplicationEventListener, ReplayListenerBus}
+import org.apache.spark.ui.SparkUI
+import org.apache.spark.util.{Clock, SystemClock}
+
+/**
+ * A History provider which reads in the history from the YARN Application Timeline Service.
+ *
+ * The service is a remote HTTP service, so failure modes are different from simple file IO.
+ *
+ * 1. Application listings are asynchronous, and made on a schedule, though
+ * they can be forced (and the schedule disabled).
+ * 2. The results are cached and can be retrieved with [[getApplications]].
+ * 3. The most recent failure of any operation is stored,
+ * The [[getLastFailure]] call will return the last exception
+ * or `None`. It is shared across threads so is primarily there for
+ * tests and basic diagnostics.
+ * 4. Listing the details of a single application in [[getAppUI()]]
+ * is synchronous and *not* cached.
+ * 5. the [[maybeCheckEndpoint()]] call performs an endpoint check as the initial
+ * binding operation of this instance. This call invokes [[TimelineQueryClient.endpointCheck()]]
+ * for better diagnostics on binding failures -particularly configuration problems.
+ * 6. Every REST call, synchronous or asynchronous, will invoke [[maybeCheckEndpoint()]] until
+ * the endpoint check eventually succeeds.
+ *
+ * If the timeline is not enabled, the API calls used by the web UI
+ * downgrade gracefully (returning empty entries), rather than fail.
+ *
+ * @param sparkConf configuration of the provider
+ */
+private[spark] class YarnHistoryProvider(sparkConf: SparkConf)
+ extends ApplicationHistoryProvider with Logging {
+
+ // import all the constants
+ import org.apache.spark.deploy.history.yarn.server.YarnHistoryProvider._
+
+ /**
+ * The configuration here is a YarnConfiguration built off the spark configuration
+ * supplied in the constructor; this operation ensures that `yarn-default.xml`
+ * and `yarn-site.xml` are pulled in. Options in the spark conf will override
+ * those in the -default and -site XML resources which are not marked as final.
+ */
+ private val yarnConf = {
+ new YarnConfiguration(SparkHadoopUtil.get.newConfiguration(sparkConf))
+ }
+
+ /**
+ * UI ACL option.
+ */
+ private val uiAclsEnabled = sparkConf.getBoolean("spark.history.ui.acls.enable", false)
+
+ /**
+ * Flag/option which triggers providing detailed diagnostics info on the property
+ * map of the history provider.
+ */
+ private val detailedInfo = sparkConf.getBoolean(OPTION_DETAILED_INFO, false)
+
+ /**
+ * The minimum interval in milliseconds between each check for event log updates.
+ */
+ val manualRefreshInterval = sparkConf.getTimeAsSeconds(OPTION_MANUAL_REFRESH_INTERVAL,
+ s"${DEFAULT_MANUAL_REFRESH_INTERVAL_SECONDS}s") * 1000
+
+ /**
+ * The interval for manual intervals.
+ */
+ val backgroundRefreshInterval = sparkConf.getTimeAsSeconds(OPTION_BACKGROUND_REFRESH_INTERVAL,
+ s"${DEFAULT_BACKGROUND_REFRESH_INTERVAL_SECONDS}s") * 1000
+
+ /**
+ * Window limit in milliseconds.
+ */
+ private val windowLimitMs = sparkConf.getTimeAsSeconds(OPTION_WINDOW_LIMIT,
+ DEFAULT_WINDOW_LIMIT) * 1000
+
+ /**
+ * Number of events to get.
+ */
+ private val eventFetchLimit = sparkConf.getLong(OPTION_EVENT_FETCH_LIMIT,
+ DEFAULT_EVENT_FETCH_LIMIT)
+
+ /**
+ * Convert the limit to an option where 0 is mapped to None.
+ */
+ private val eventFetchOption: Option[Long] =
+ if (eventFetchLimit > 0) Some(eventFetchLimit) else None
+
+ /**
+ * Clock. Subclasses may override for testing.
+ */
+ protected var clock: Clock = new SystemClock()
+
+ /**
+ * YARN client used to list running apps and so infer which of the incomplete apps have really
+ * finished.
+ */
+ private var yarnClient: YarnClient = _
+
+ /**
+ * Are liveness checks enabled?
+ * That is, is the option [[OPTION_YARN_LIVENESS_CHECKS]] set?
+ */
+ val _livenessChecksEnabled = sparkConf.getBoolean(OPTION_YARN_LIVENESS_CHECKS, true)
+
+ /**
+ * Start time in milliseconds.
+ */
+ val serviceStartTime = clock.getTimeMillis()
+
+ /**
+ * Timeline endpoint URI.
+ */
+ protected val timelineEndpoint = createTimelineEndpoint()
+
+ /**
+ * Create a [[TimelineQueryClient]] instance to talk to the timeline service running
+ * at [[timelineEndpoint]].
+ *
+ */
+ protected val timelineQueryClient = {
+ createTimelineQueryClient()
+ }
+
+ /**
+ * Override point: create the timeline endpoint.
+ *
+ * @return a URI to the timeline web service
+ */
+ protected def createTimelineEndpoint(): URI = {
+ getTimelineEndpoint(yarnConf)
+ }
+
+ /**
+ * Create the timeline query client.
+ *
+ * This is called during instance creation; tests may override this
+ * @return a timeline query client for use for the duration
+ * of this instance
+ */
+ protected def createTimelineQueryClient(): TimelineQueryClient = {
+ new TimelineQueryClient(timelineEndpoint, yarnConf, JerseyBinding.createClientConfig())
+ }
+
+ /**
+ * The empty listing, with a timestamp to indicate that the listing
+ * has never taken place.
+ */
+ private val EmptyListing = new ApplicationListingResults(0, Nil, None)
+
+ /**
+ * List of applications. Initial result is empty.
+ */
+ private var applications: ApplicationListingResults = EmptyListing
+
+ /**
+ * Last exception seen and when.
+ */
+ protected var lastFailureCause: Option[(Throwable, Date)] = None
+
+ /**
+ * Flag to indicate an endpoint diagnostics probe should take place.
+ */
+ protected val endpointCheckExecuted = new AtomicBoolean(false)
+
+ /**
+ * Flag set to indicate refresh is in progress.
+ */
+ private val _refreshInProgress = new AtomicBoolean(false)
+
+ /**
+ * How long did that last refresh take?
+ */
+ private val _lastRefreshDuration = new AtomicLong(0)
+
+ /**
+ * Enabled flag.
+ */
+ private val _enabled = timelineServiceEnabled(yarnConf)
+
+ /**
+ * Atomic boolean used to signal to the refresh thread that it
+ * must exit its loop.
+ */
+ private val stopRefresh = new AtomicBoolean(false)
+
+ /**
+ * Refresher thread.
+ */
+ private[yarn] val refresher = new Refresher()
+
+
+ /**
+ * Time in milliseconds *before* the last application report to set the window
+ * start on the next query. Keeping this earlier than the last report ensures that
+ * changes to that application are picked up.
+ */
+ private val StartWindowOffsetMillis = 1000
+
+ /**
+ * Component Metrics. Accessible for the benefit of tests.
+ */
+ val metrics = new YarnHistoryProviderMetrics(this)
+
+ /**
+ * Initialize the provider.
+ */
+ init()
+
+ /**
+ * Check the configuration and log whether or not it is enabled;
+ * if it is enabled then the refresh thread starts.
+ *
+ * Starting threads in a constructor is considered awful practice as
+ * it can leak reference and cause chaos and confusion with subclasses.
+ * But in the absence of some start() method for history providers, it's
+ * all there is.
+ */
+ private def init(): Unit = {
+ if (!enabled) {
+ logError(TEXT_SERVICE_DISABLED)
+ } else {
+ logInfo(TEXT_SERVICE_ENABLED)
+ logInfo(KEY_SERVICE_URL + ": " + timelineEndpoint)
+ logDebug(sparkConf.toDebugString)
+ // get the thread time
+ logInfo(s"Manual refresh interval $manualRefreshInterval milliseconds")
+ validateInterval(OPTION_MANUAL_REFRESH_INTERVAL, manualRefreshInterval, 0)
+ // check the background refresh interval if there is one
+ if (backgroundRefreshInterval > 0) {
+ validateInterval(OPTION_BACKGROUND_REFRESH_INTERVAL, backgroundRefreshInterval,
+ MIN_BACKGROUND_REFRESH_INTERVAL)
+ logInfo(s"Background refresh interval $backgroundRefreshInterval milliseconds")
+ } else {
+ logInfo(s"Background refresh interval=0: manual refreshes only")
+ }
+ initYarnClient()
+ startRefreshThread()
+ }
+ }
+
+ /**
+ * Verify that a time interval is at or above the minimum allowed.
+ *
+ * @param prop property for use in exception text
+ * @param interval interval to valide
+ * @param min minimum allowed value
+ */
+ private def validateInterval(prop: String, interval: Long, min: Long): Unit = {
+ if (interval < min) {
+ throw new IllegalArgumentException(TEXT_INVALID_UPDATE_INTERVAL +
+ prop + s": ${interval / 1000}; minimum allowed = ${min / 1000}")
+ }
+ }
+
+ /**
+ * Stop the service. After this point operations will fail.
+ */
+ override def stop(): Unit = {
+ logDebug(s"Stopping $this")
+ // attempt to stop the refresh thread
+ if (enabled) {
+ if (!stopRefreshThread()) {
+ closeQueryClient()
+ ServiceOperations.stop(yarnClient)
+ }
+ }
+ }
+
+ /**
+ * Close the query client.
+ */
+ private def closeQueryClient(): Unit = {
+ logDebug("Stopping Timeline client")
+ timelineQueryClient.close()
+ }
+
+ /**
+ * Is the timeline service (and therefore this provider) enabled.
+ * (override point for tests)?
+ *
+ * Important: this is called during construction, so test-time subclasses
+ * will be invoked before their own construction has taken place.
+ * Code appropriately.
+ *
+ * @return true if the provider/YARN configuration enables the timeline
+ * service.
+ */
+ def enabled: Boolean = {
+ _enabled
+ }
+
+ /**
+ * Are liveness checks enabled?
+ * @return true if the liveness checks on the service endpoint take place.
+ */
+ def livenessChecksEnabled: Boolean = {
+ _livenessChecksEnabled
+ }
+
+ /**
+ * Is a refresh in progress?
+ *
+ * @return flag to indicate a refresh is in progress
+ */
+ def refreshInProgress: Boolean = {
+ _refreshInProgress.get()
+ }
+
+ /**
+ * The duration of last refresh.
+ * @return a duration in millis; 0 if no refresh has taken place
+ */
+ def lastRefreshDuration: Long = {
+ _lastRefreshDuration.get()
+ }
+
+ /**
+ * Get the timeline query client. Used internally to ease testing
+ * @return the client.
+ */
+ def getTimelineQueryClient: TimelineQueryClient = {
+ timelineQueryClient
+ }
+
+ /**
+ * Set the last exception.
+ * @param ex exception seen
+ */
+ private def setLastFailure(ex: Throwable): Unit = {
+ setLastFailure(ex, now())
+ }
+
+ /**
+ * Set the last exception.
+ *
+ * @param ex exception seen
+ * @param timestamp the timestamp of the failure
+ */
+ private def setLastFailure(ex: Throwable, timestamp: Long): Unit = {
+ synchronized {
+ lastFailureCause = Some((ex, new Date(timestamp)))
+ }
+ }
+
+ /**
+ * Reset the failure info.
+ */
+ private def resetLastFailure(): Unit = {
+ synchronized {
+ lastFailureCause = None
+ }
+ }
+
+ /**
+ * Get the last exception.
+ *
+ * @return the last exception or null
+ */
+ def getLastFailure: Option[(Throwable, Date)] = {
+ synchronized {
+ lastFailureCause
+ }
+ }
+
+ /**
+ * Thread safe accessor to the application list.
+ *
+ * @return a list of applications
+ */
+ def getApplications: ApplicationListingResults = {
+ synchronized {
+ applications
+ }
+ }
+
+ /**
+ * Thread safe call to update the application results.
+ *
+ * @param newVal new value
+ */
+ private def setApplications(newVal: ApplicationListingResults): Unit = {
+ synchronized {
+ applications = newVal
+ }
+ }
+
+ /**
+ * Reachability check to call before any other operation is attempted.
+ * This is atomic, using the `shouldCheckEndpoint` flag to check.
+ * If the endpoint check failes then the
+ * `endpointCheckExecuted` flag is reset to false and an exception thrown.
+ *
+ * @return true if the check took place
+ */
+ protected def maybeCheckEndpoint(): Boolean = {
+ if (!endpointCheckExecuted.getAndSet(true)) {
+ val client = getTimelineQueryClient
+ try {
+ client.endpointCheck()
+ true
+ } catch {
+ case e: Exception =>
+ // failure
+ logWarning(s"Endpoint check of $client failed", e)
+ setLastFailure(e)
+ // reset probe so another caller may attempt it.
+ endpointCheckExecuted.set(false)
+ // propagate the failure
+ throw e
+ }
+ } else {
+ false
+ }
+ }
+
+ /**
+ * Query for the endpoint check being successful.
+ *
+ * Note: also true if the check is in progress.
+ *
+ * @return true if the check has succeeded, or it is actually ongoing (it may fail)
+ */
+ def endpointCheckSuccess(): Boolean = {
+ endpointCheckExecuted.get()
+ }
+
+ /**
+ * Start the refresh thread with the given interval.
+ *
+ * When this thread exits, it will close the `timelineQueryClient`
+ * instance
+ */
+ def startRefreshThread(): Unit = {
+ logInfo(s"Starting timeline refresh thread")
+ val thread = new Thread(refresher, s"YarnHistoryProvider Refresher")
+ thread.setDaemon(true)
+ refresher.start(thread)
+ }
+
+ /**
+ * Stop the refresh thread if there is one.
+ *
+ * This does not guarantee an immediate halt to the thread.
+ *
+ * @return true if there was a refresh thread to stop
+ */
+ private def stopRefreshThread(): Boolean = {
+ refresher.stopRefresher()
+ }
+
+ /**
+ * Probe for the refresh thread running.
+ *
+ * @return true if the refresh thread has been created and is still alive
+ */
+ def isRefreshThreadRunning: Boolean = {
+ refresher.isRunning
+ }
+
+ /**
+ * Get the number of times an attempt was made to refresh the listing.
+ * This is incremented on every operation, irrespective of the outcome.
+ *
+ * @return the current counter of refresh attempts.
+ */
+ def refreshCount: Long = {
+ metrics.refreshCount.getCount
+ }
+
+ /**
+ * Get number of failed refreshes.
+ *
+ * Invariant: always equal to or less than [[refreshCount]]
+ *
+ * @return the number of times refreshes failed
+ */
+ def refreshFailedCount: Long = {
+ metrics.refreshFailedCount.getCount()
+ }
+
+ /**
+ * List applications.
+ *
+ * If the timeline is not enabled, returns `emptyListing`
+ * @return the result of the last successful listing operation,
+ * or a listing with no history events if there has been a failure
+ */
+ def listApplications(
+ limit: Option[Long] = None,
+ windowStart: Option[Long] = None,
+ windowEnd: Option[Long] = None): ApplicationListingResults = {
+ if (!enabled) {
+ // Timeline is disabled: return the empty listing
+ return EmptyListing
+ }
+ try {
+ maybeCheckEndpoint()
+ val client = getTimelineQueryClient
+ logInfo(s"getListing from: $client")
+ // get the timestamp after any endpoint check
+ val timestamp = now()
+ // list the entities, excluding the events -this is critical for performance reasons
+ val timelineEntities = client.listEntities(SPARK_EVENT_ENTITY_TYPE,
+ windowStart = windowStart,
+ windowEnd = windowEnd,
+ limit = limit,
+ fields = Seq(PRIMARY_FILTERS, OTHER_INFO)
+ )
+
+ // build one history info entry for each entity in the least (implicitly, one
+ // attempt per history info entry
+ val umergedHistory = timelineEntities.flatMap { en =>
+ try {
+ val historyInfo = toApplicationHistoryInfo(en)
+ logDebug(s"${describeApplicationHistoryInfo(historyInfo)}")
+ Some(historyInfo)
+ } catch {
+ case e: Exception =>
+ logWarning(s"Failed to parse entity. ${YarnTimelineUtils.describeEntity(en)}", e)
+ // skip this result
+ None
+ }
+ }
+ // merge the results so that multiple attempts are combined into
+ // single history entries
+ val histories = combineResults(Nil, umergedHistory)
+ val incomplete = countIncompleteApplications(histories)
+ logInfo(s"Found ${histories.size} application(s): " +
+ s"${histories.size - incomplete} complete and $incomplete incomplete")
+ new ApplicationListingResults(timestamp, histories, None)
+ } catch {
+ case e: IOException =>
+ logWarning(s"Failed to list entities from $timelineEndpoint", e)
+ new ApplicationListingResults(now(), Nil, Some(e))
+ }
+ }
+
+ /**
+ * List applications.
+ *
+ * Also updates the cached values of the listing/last failure, depending
+ * upon the outcome.
+ *
+ * If the timeline is not enabled, returns an empty list.
+ *
+ * @param startup a flag to indicate this is the startup retrieval with different window policy
+ * @return List of all known applications.
+ */
+ def listAndCacheApplications(startup: Boolean): ApplicationListingResults = {
+ metrics.refreshCount.inc()
+ _refreshInProgress.set(true)
+ val history = getApplications.applications
+
+ val refreshStartTime = now()
+ metrics.time(metrics.refreshDuration) {
+ try {
+ // work out the start of the new window
+ val nextWindowStart = if (startup) {
+ None
+ } else {
+ findStartOfWindow(history) map { app =>
+ // inclusive on the one retrieved last time.
+ // Why? we need to include the oldest incomplete entry in our range
+ val inclusiveWindow = startTime(app) - StartWindowOffsetMillis
+ // sanity check on window size
+ val earliestWindow = if (windowLimitMs > 0) refreshStartTime - windowLimitMs else 0
+ Math.max(earliestWindow, inclusiveWindow)
+ }
+ }
+ val results = listApplications(windowStart = nextWindowStart)
+ synchronized {
+ if (results.succeeded) {
+ // on a success, the existing application list is merged
+ // creating a new aggregate application list
+ logDebug(s"Listed application count: ${results.size}")
+ val merged = combineResults(history, results.applications)
+ logDebug(s"Existing count: ${history.size}; merged = ${merged.size} ")
+ val updated = if (livenessChecksEnabled) {
+ updateAppsFromYARN(merged)
+ } else {
+ merged
+ }
+ val sorted = sortApplicationsByStartTime(updated)
+ // and a final result
+ setApplications(new ApplicationListingResults(results.timestamp, sorted, None))
+ resetLastFailure()
+ } else {
+ // on a failure, the failure cause is updated
+ setLastFailure(results.failureCause.get, results.timestamp)
+ // and the failure counter
+ metrics.refreshFailedCount.inc()
+ }
+ }
+ results
+ } finally {
+ _refreshInProgress.set(false)
+ _lastRefreshDuration.set(now() - refreshStartTime)
+ }
+ }
+ }
+
+ /**
+ * List applications.
+ *
+ * If the timeline is not enabled, returns an empty list
+ * @return List of all known applications.
+ */
+ override def getListing(): Seq[TimelineApplicationHistoryInfo] = {
+ // get the current list
+ val listing = getApplications.applications
+ // and queue another refresh
+ triggerRefresh()
+ listing
+ }
+
+ /**
+ * Trigger a refresh.
+ */
+ private[yarn] def triggerRefresh(): Unit = {
+ refresher.refresh(now())
+ }
+
+ /**
+ * Return the current time.
+ * @return the time in milliseconds.
+ */
+ private[yarn] def now(): Long = {
+ clock.getTimeMillis()
+ }
+
+ /**
+ * Get the last refresh attempt (Which may or may not have been successful).
+ *
+ * @return the last refresh time
+ */
+ def lastRefreshAttemptTime: Long = {
+ refresher.lastRefreshAttemptTime
+ }
+
+ /**
+ * Look up the timeline entity.
+ *
+ * @param entityId application ID
+ * @return the entity associated with the given application
+ * @throws FileNotFoundException if no entry was found
+ */
+ def getTimelineEntity(entityId: String): TimelineEntity = {
+ logDebug(s"GetTimelineEntity $entityId")
+ metrics.time(metrics.attemptFetchDuration){
+ maybeCheckEndpoint()
+ getTimelineQueryClient.getEntity(SPARK_EVENT_ENTITY_TYPE, entityId)
+ }
+ }
+
+ /**
+ * Returns the Spark UI for a specific application.
+ *
+ * If the timeline is not enabled, returns `None`
+ *
+ * @param appId The application ID.
+ * @param attemptId The application attempt ID (or `None` if there is no attempt ID).
+ * @return The application's UI, or `None` if application is not found.
+ */
+ override def getAppUI(appId: String, attemptId: Option[String]): Option[SparkUI] = {
+
+ logInfo(s"Request UI with appId $appId attempt $attemptId")
+ if (!enabled) {
+ // Timeline is disabled: return nothing
+ return None
+ }
+ maybeCheckEndpoint()
+ metrics.attemptLoadCount.inc()
+ if (attemptId.isEmpty) {
+ // empty attempts are rejected
+ metrics.attemptLoadFailureCount.inc()
+ return None
+ }
+
+ // the URL generation/linking code in the History Server appears fairly brittle,
+ // so provide as diagnostics information in the logs to help understand what
+ // is happending
+ val (foundApp, attempt, attempts) = getApplications.lookupAttempt(appId, attemptId)
+ if (foundApp.isEmpty) {
+ log.error(s"Application $appId not found")
+ metrics.attemptLoadFailureCount.inc()
+ return None
+ }
+ val attemptInfo = attempt.getOrElse {
+ log.error(s"Attempt $attemptId not found under application $appId")
+ log.error(s"${attempts.length} attempts: " + attempts.mkString("[", ", ", "]"))
+ metrics.attemptLoadFailureCount.inc()
+ return None
+ }
+
+ val entityId = attemptInfo.entityId
+ val appInfo = foundApp.get
+
+ metrics.time(metrics.attemptLoadDuration) {
+ try {
+ val attemptEntity = getTimelineEntity(entityId)
+ if (log.isDebugEnabled) {
+ logDebug(describeEntity(attemptEntity))
+ }
+ val bus = new ReplayListenerBus()
+ val appListener = new ApplicationEventListener()
+ bus.addListener(appListener)
+ val conf = this.sparkConf.clone()
+ val attemptURI = HistoryServer.getAttemptURI(appId, attemptId)
+ val name = appInfo.name
+ val ui = SparkUI.createHistoryUI(conf,
+ bus,
+ new SecurityManager(conf),
+ name,
+ attemptURI,
+ attemptInfo.startTime)
+ ui.setAppId(appId)
+ logInfo(s"Building Application UI $name attempt $attemptId under ${ui.basePath}")
+
+ // replay all the events
+ val events = attemptEntity.getEvents.asScala
+ logInfo(s"App $appId history contains ${events.length} events")
+
+ metrics.time(metrics.attemptReplayDuration) {
+ events.reverse.foreach { event =>
+ val sparkEvent = toSparkEvent(event)
+ bus.postToAll(sparkEvent)
+ }
+ }
+
+ ui.getSecurityManager.setAcls(uiAclsEnabled)
+ // make sure to set admin acls before view acls so they are properly picked up
+ ui.getSecurityManager.setAdminAcls(appListener.adminAcls.getOrElse(""))
+ ui.getSecurityManager.setViewAcls(appListener.sparkUser.getOrElse("historyService
+ */
+ override def afterEach(): Unit = {
+ describe("Teardown of history server, timeline client and history service")
+ stopHistoryService(historyService)
+ historyService = null
+ ServiceOperations.stopQuietly(_applicationHistoryServer)
+ _applicationHistoryServer = null
+ ServiceOperations.stopQuietly(_timelineClient)
+ _timelineClient = null
+ super.afterEach()
+ }
+
+ /**
+ * Stop a history service. This includes flushing its queue,
+ * blocking until that queue has been flushed and closed, then
+ * stopping the YARN service.
+ * @param hservice history service to stop
+ */
+ def stopHistoryService(hservice: YarnHistoryService): Unit = {
+ if (hservice != null && hservice.serviceState == YarnHistoryService.StartedState) {
+ flushHistoryServiceToSuccess()
+ hservice.stop()
+ awaitServiceThreadStopped(hservice, TEST_STARTUP_DELAY, false)
+ }
+ }
+
+ /**
+ * Add an action to execute on failures (if the test runs it
+ * @param action action to execute
+ */
+ def addFailureAction(action: () => Unit) : Unit = {
+ failureActions += action
+ }
+
+ /**
+ * Execute all the failure actions in order.
+ */
+ def executeFailureActions(): Unit = {
+ if (failureActions.nonEmpty) {
+ logError("== Executing failure actions ==")
+ }
+ failureActions.foreach{ action =>
+ try {
+ action()
+ } catch {
+ case _ : Exception =>
+ }
+ }
+ }
+
+ /**
+ * Failure action to log history service details at INFO
+ */
+ def dumpYarnHistoryService(): Unit = {
+ if (historyService != null) {
+ logError(s"-- History Service --\n$historyService")
+ }
+ }
+
+ /**
+ * Curryable Failure action to dump provider state
+ * @param provider the provider
+ */
+ def dumpProviderState(provider: YarnHistoryProvider)(): Unit = {
+ logError(s"-- Provider --\n$provider")
+ val results = provider.getApplications
+ results.applications.foreach{
+ app =>
+ logError(s" $app")
+ }
+ results.failureCause.foreach{ e =>
+ logError("Failed", e)
+ }
+ }
+
+ /**
+ * Curryable Failure action to log all timeline entities
+ * @param provider the provider bonded to the endpoint
+ */
+ def dumpTimelineEntities(provider: YarnHistoryProvider)(): Unit = {
+ logError("-- Dumping timeline entities --")
+ val entities = provider.getTimelineQueryClient
+ .listEntities(YarnHistoryService.SPARK_EVENT_ENTITY_TYPE)
+ entities.foreach{ e =>
+ logError(describeEntity(e))
+ }
+ }
+
+ /**
+ * Create a SPNEGO-enabled URL Connector.
+ * Picks up the hadoop configuration from `sc`, so the context
+ * must be live/non-null
+ * @return a URL connector for issuing HTTP requests
+ */
+ protected def createUrlConnector(): SpnegoUrlConnector = {
+ val hadoopConfiguration = sc.hadoopConfiguration
+ createUrlConnector(hadoopConfiguration)
+ }
+
+ /**
+ * Create a SPNEGO-enabled URL Connector.
+ * @param hadoopConfiguration the configuration to use
+ * @return a URL connector for issuing HTTP requests
+ */
+ def createUrlConnector(hadoopConfiguration: Configuration): SpnegoUrlConnector = {
+ SpnegoUrlConnector.newInstance(hadoopConfiguration, new Token)
+ }
+
+ /**
+ * Create the client and the app server
+ * @param conf the hadoop configuration
+ */
+ protected def startTimelineClientAndAHS(conf: Configuration): Unit = {
+ ServiceOperations.stopQuietly(_applicationHistoryServer)
+ ServiceOperations.stopQuietly(_timelineClient)
+ _timelineClient = TimelineClient.createTimelineClient()
+ _timelineClient.init(conf)
+ _timelineClient.start()
+ _applicationHistoryServer = new ApplicationHistoryServer()
+ _applicationHistoryServer.init(_timelineClient.getConfig)
+ _applicationHistoryServer.start()
+ // Wait for AHS to come up
+ val endpoint = YarnTimelineUtils.timelineWebappUri(conf, "")
+ awaitURL(endpoint.toURL, TEST_STARTUP_DELAY)
+ }
+
+ protected def createTimelineQueryClient(): TimelineQueryClient = {
+ new TimelineQueryClient(historyService.timelineWebappAddress,
+ sc.hadoopConfiguration, createClientConfig())
+ }
+
+ /**
+ * Put a timeline entity to the timeline client; this is expected
+ * to eventually make it to the history server
+ * @param entity entity to put
+ * @return the response
+ */
+ def putTimelineEntity(entity: TimelineEntity): TimelinePutResponse = {
+ assertNotNull(_timelineClient, "timelineClient")
+ _timelineClient.putEntities(entity)
+ }
+
+ /**
+ * Marshall and post a spark event to the timeline; return the outcome
+ * @param sparkEvt event
+ * @param time event time
+ * @return a triple of the wrapped event, marshalled entity and the response
+ */
+ protected def postEvent(sparkEvt: SparkListenerEvent, time: Long):
+ (TimelineEvent, TimelineEntity, TimelinePutResponse) = {
+ val event = toTimelineEvent(sparkEvt, time).get
+ val entity = newEntity(time)
+ entity.addEvent(event)
+ val response = putTimelineEntity(entity)
+ val description = describePutResponse(response)
+ logInfo(s"response: $description")
+ assert(response.getErrors.isEmpty, s"errors in response: $description")
+ (event, entity, response)
+ }
+
+ /**
+ * flush the history service of its queue, await it to complete,
+ * then assert that there were no failures
+ */
+ protected def flushHistoryServiceToSuccess(): Unit = {
+ flushHistoryServiceToSuccess(historyService)
+ }
+
+ /**
+ * Flush a history service to success
+ * @param history service to flush
+ * @param delay time to wait for an empty queue
+ */
+ def flushHistoryServiceToSuccess(
+ history: YarnHistoryService,
+ delay: Int = TEST_STARTUP_DELAY): Unit = {
+ assertNotNull(history, "null history queue")
+ historyService.asyncFlush()
+ awaitEmptyQueue(history, delay)
+ assert(0 === history.postFailures, s"Post failure count: $history")
+ }
+
+ /**
+ * Create a history provider instance, following the same process
+ * as the history web UI itself: querying the configuration for the
+ * provider and falling back to the [[FsHistoryProvider]]. If
+ * that falback does take place, however, and assertion is raised.
+ * @param conf configuration
+ * @return the instance
+ */
+ protected def createHistoryProvider(conf: SparkConf): YarnHistoryProvider = {
+ val providerName = conf.getOption("spark.history.provider")
+ .getOrElse(classOf[FsHistoryProvider].getName())
+ val provider = Utils.classForName(providerName)
+ .getConstructor(classOf[SparkConf])
+ .newInstance(conf)
+ .asInstanceOf[ApplicationHistoryProvider]
+ assert(provider.isInstanceOf[YarnHistoryProvider],
+ s"Instantiated $providerName to get $provider")
+
+ provider.asInstanceOf[YarnHistoryProvider]
+ }
+
+ /**
+ * Crete a history server and maching provider, execute the
+ * probe against it. After the probe completes, the history server
+ * is stopped.
+ * @param probe probe to run
+ */
+ def webUITest(name: String, probe: (URL, YarnHistoryProvider) => Unit): Unit = {
+ val (_, server, webUI, provider) = createHistoryServer(findPort())
+ try {
+ sparkHistoryServer = server
+ server.bind()
+ describe(name)
+ probe(webUI, provider)
+ } catch {
+ case ex: Exception =>
+ executeFailureActions()
+ throw ex
+ } finally {
+ describe("stopping history service")
+ Utils.tryLogNonFatalError {
+ server.stop()
+ }
+ sparkHistoryServer = null
+ }
+ }
+
+ /**
+ * Probe the empty web UI for not having any completed apps; expect
+ * a text/html response with specific text and history provider configuration
+ * elements.
+ * @param webUI web UI
+ * @param provider provider
+ */
+ def probeEmptyWebUI(webUI: URL, provider: YarnHistoryProvider): String = {
+ val body: String = getHtmlPage(webUI,
+ "provider.getAppUI
+ */
+ def getAppUI(appid: String, attemptId: Option[String] = None): Option[SparkUI] = {
+ logDebug(s"GET appID =$appid attemptId=$attemptId" )
+ provider.getAppUI(appid, attemptId)
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/integration/YarnHistoryProviderWindowSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/integration/YarnHistoryProviderWindowSuite.scala
new file mode 100644
index 000000000000..7c9c9efcc7a7
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/integration/YarnHistoryProviderWindowSuite.scala
@@ -0,0 +1,145 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.integration
+
+import org.apache.hadoop.yarn.api.records.YarnApplicationState
+
+import org.apache.spark.deploy.history.yarn.YarnHistoryService
+import org.apache.spark.deploy.history.yarn.server.YarnProviderUtils._
+import org.apache.spark.deploy.history.yarn.testtools.{HistoryServiceNotListeningToSparkContext, TimelineSingleEntryBatchSize}
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+import org.apache.spark.scheduler.cluster.StubApplicationAttemptId
+import org.apache.spark.util.Utils
+
+/**
+ * check windowed providder.
+ *
+ * More than one history service is started here, each publishing their own events, with
+ * their own app ID. For this to work they are set up to not listen to context events.
+ */
+class YarnHistoryProviderWindowSuite
+ extends AbstractHistoryIntegrationTests
+ with HistoryServiceNotListeningToSparkContext
+ with TimelineSingleEntryBatchSize {
+ val minute = 60000
+ val start1Time = minute
+ val start2Time = start1Time + minute
+ val appReport1 = stubApplicationReport(1, 0, 1, YarnApplicationState.RUNNING, start1Time, 0)
+ val appReport2 = stubApplicationReport(2, 0, 1, YarnApplicationState.RUNNING, start2Time, 0)
+
+ val appId1 = appReport1.getApplicationId.toString
+ val appId2 = appReport2.getApplicationId.toString
+ val user = Utils.getCurrentUserName()
+
+ test("YarnHistoryProviderWindow") {
+ describe("Windowed publishing across apps")
+ var history2: YarnHistoryService = null
+ var provider: TimeManagedHistoryProvider = null
+ try {
+ logDebug("Start application 1")
+ val expectedAppId1 = appReport1.getApplicationId
+ historyService = startHistoryService(sc, expectedAppId1,
+ Some(appReport1.getCurrentApplicationAttemptId))
+ assert(!historyService.listening, s"listening $historyService")
+ assert(historyService.bondedToATS, s"not bonded to ATS: $historyService")
+ // post in an app start
+ val start1 = appStartEvent(start1Time, appId1, user, Some("1"))
+ enqueue(start1)
+ flushHistoryServiceToSuccess(historyService)
+
+ describe("application 2")
+ // the second application starts then stops after the first one
+ val applicationId2 = appReport2.getApplicationId
+ val attemptId2 = new StubApplicationAttemptId(applicationId2, 2)
+ val expectedAppId2 = applicationId2.toString
+ history2 = startHistoryService(sc, applicationId2,
+ Some(appReport2.getCurrentApplicationAttemptId))
+
+
+ val start2 = appStartEvent(start2Time, appId2, user, Some("1"))
+ history2.enqueue(start2)
+ val end2Time = start2Time + minute
+ val end2 = appStopEvent(end2Time)
+ history2.enqueue(end2)
+ // stop the second application
+ history2.stop()
+ flushHistoryServiceToSuccess(history2)
+ history2 = null
+
+ // here there is one incomplete application, and a completed one
+ // which started and stopped after the incomplete one started
+ provider = new TimeManagedHistoryProvider(sc.conf, end2Time, minute)
+ provider.setRunningApplications(List(appReport1, appReport2))
+ addFailureAction(dumpProviderState(provider))
+ addFailureAction(dumpTimelineEntities(provider))
+
+ // now read it in via history provider
+ describe("read in listing")
+
+ val listing1 = awaitApplicationListingSize(provider, 2, TEST_STARTUP_DELAY)
+ logInfo(s"Listing 1: $listing1")
+ assertAppCompleted(lookupApplication(listing1, expectedAppId2),
+ s"app2 ID $expectedAppId2, in listing1 $listing1")
+ val applicationInfo1_1 = lookupApplication(listing1, expectedAppId1)
+ assert(!isCompleted(applicationInfo1_1), s"$applicationInfo1_1 completed in L1 $listing1")
+
+ describe("stop application 1")
+ val end3Time = provider.tick()
+ val end3 = appStopEvent(end3Time)
+ historyService.enqueue(end3)
+ historyService.stop()
+ flushHistoryServiceToSuccess()
+
+ // move time forwards
+ provider.incrementTime(5 * minute)
+ // Now await a refresh
+ describe("read in listing #2")
+
+ awaitRefreshExecuted(provider, true, TEST_STARTUP_DELAY)
+ awaitRefreshExecuted(provider, true, TEST_STARTUP_DELAY)
+ awaitRefreshExecuted(provider, true, TEST_STARTUP_DELAY)
+
+ logDebug("Refreshes executed; extracting application listing")
+ val allApps = provider.listApplications()
+ logInfo(s"allApps : ${allApps.applications}")
+
+ // get a new listing
+ val listing2 = provider.getListing()
+ logInfo(s"Listing 2: $listing2")
+ // which had better be updated or there are refresh problems
+ assert(listing1 !== listing2, s"updated listing was unchanged from $provider")
+ // get the updated value and expect it to be complete
+ assertAppCompleted(lookupApplication(listing2, expectedAppId1), s"app1 in L2 $listing2")
+ assertAppCompleted(lookupApplication(listing2, expectedAppId1), s"app2 in L2 $listing2")
+ provider.stop()
+ } catch {
+ case ex: Exception =>
+ executeFailureActions()
+ throw ex
+ } finally {
+ describe("teardown")
+ if (history2 != null) {
+ history2.stop()
+ }
+ if (provider != null) {
+ provider.stop()
+ }
+ }
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/integration/YarnProviderUtilsSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/integration/YarnProviderUtilsSuite.scala
new file mode 100644
index 000000000000..b7d6601a791d
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/integration/YarnProviderUtilsSuite.scala
@@ -0,0 +1,401 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.integration
+
+import org.scalatest.Matchers
+
+import org.apache.spark.deploy.history.yarn.YarnHistoryService._
+import org.apache.spark.deploy.history.yarn.YarnTimelineUtils._
+import org.apache.spark.deploy.history.yarn.server.{TimelineApplicationHistoryInfo, TimelineApplicationAttemptInfo}
+import org.apache.spark.deploy.history.yarn.server.YarnProviderUtils._
+import org.apache.spark.deploy.history.yarn.testtools.ExtraAssertions
+import org.apache.spark.scheduler.cluster.{StubApplicationAttemptId, StubApplicationId}
+import org.apache.spark.{Logging, SparkFunSuite}
+
+/**
+ * Test of utility methods in [[org.apache.spark.deploy.history.yarn.server.YarnProviderUtils]]
+ */
+class YarnProviderUtilsSuite extends SparkFunSuite with Logging
+ with ExtraAssertions with Matchers {
+
+ def historyInfo(
+ id: String,
+ started: Long,
+ ended: Long,
+ complete: Boolean): TimelineApplicationHistoryInfo = {
+ historyInfo(id, Some(id), started, ended, ended, complete )
+ }
+
+ def historyInfo(
+ appId: String,
+ attemptId: Option[String],
+ started: Long,
+ ended: Long,
+ updated: Long,
+ complete: Boolean): TimelineApplicationHistoryInfo = {
+ val updated = Math.max(started, ended)
+ val attempt = new TimelineApplicationAttemptInfo(attemptId,
+ started, ended, updated, "user", complete , attemptId.get, attemptId)
+ new TimelineApplicationHistoryInfo(appId, appId, List(attempt))
+ }
+
+ def historyInfo(
+ old: TimelineApplicationHistoryInfo,
+ attempts: List[TimelineApplicationAttemptInfo]): TimelineApplicationHistoryInfo = {
+ new TimelineApplicationHistoryInfo(old.id, old.name, attempts)
+ }
+
+ val yarnAppId = new StubApplicationId(5, 0)
+ val yarnAttemptId = new StubApplicationAttemptId(yarnAppId, 1)
+
+ val h12 = historyInfo("h12", 1, 2, true)
+ val h22 = historyInfo("h22", 2, 2, true)
+ val i20 = historyInfo("i20", 2, 0, false)
+ val i30 = historyInfo("i30", 3, 0, false)
+ val h33 = historyInfo("h30", 3, 3, true)
+ val h44 = historyInfo("h44", 4, 4, true)
+ val iA10_incomplete = historyInfo("iA", 1, 0, false)
+ val iA11_completed = historyInfo("iA", 1, 1, true)
+ val a1_attempt_1 = historyInfo("iA", Some("attempt_1"), 100, 102, 102, false)
+ val a1_attempt_2 = historyInfo("iA", Some("attempt_2"), 200, 202, 202, true)
+ val none_incomplete = new TimelineApplicationAttemptInfo(None, 100, 0, 102, "spark", false,
+ "001", None)
+ val none_completed = new TimelineApplicationAttemptInfo(None, 200, 202, 202, "spark", true,
+ "001", None)
+ val none_completed_orig_time = new TimelineApplicationAttemptInfo(None, 100, 0, 102, "spark",
+ true, "001", None)
+ // app attempt started @ 100, updated @102, version 1
+ val attempt_1_1 = new TimelineApplicationAttemptInfo(Some("attempt_1_1"),
+ 100, 0, 102, "spark", false, "001", None, 1)
+ // attempt 1.1 updated at time = 102; no version field
+ val attempt_1_1_updated = new TimelineApplicationAttemptInfo(Some("attempt_1_1"),
+ 100, 0, 150, "spark", false, "001", None)
+ // attempt 1.1 with the version field updated to 2; it should always be newer
+ val attempt_1_1_updated_version = new TimelineApplicationAttemptInfo(Some("attempt_1_1"),
+ 100, 0, 300, "spark", false, "001", None, 2)
+ val attempt_1_1_completed_v4 = new TimelineApplicationAttemptInfo(Some("attempt_1_1"),
+ 100, 400, 400, "spark", true, "001", None, 4)
+ val attempt_1_2 = new TimelineApplicationAttemptInfo(Some("attempt_1_2"), 200, 202, 202,
+ "spark", true, "001", None, 3)
+
+ test("timeShort") {
+ assert("unset" === timeShort(0, "unset"))
+ assert("unset" !== timeShort(System.currentTimeMillis(), "unset"))
+ }
+
+ test("findOldest") {
+ assert(Some(h12) === findOldestApplication(List(h12, h22, i20)))
+ }
+
+ test("findOldest-2") {
+ assert(Some(h22) === findOldestApplication(List(h44, h22, i20)))
+ }
+
+ test("findOldest-3") {
+ assert(Some(i20) === findOldestApplication(List(h44, h33, i20)))
+ }
+
+ test("findOldest-4") {
+ assert(None === findOldestApplication(Nil))
+ }
+
+ test("findIncomplete") {
+ assert(List(i20, i30) === findIncompleteApplications(List(h44, i20, i30, h33)))
+ }
+
+ test("findIncomplete-2") {
+ assert(Nil === findIncompleteApplications(Nil))
+ }
+
+ test("findIncomplete-3") {
+ assert(Nil === findIncompleteApplications(List(h44, h33)))
+ }
+
+ test("countIncomplete") {
+ assert(2 === countIncompleteApplications(List(h44, i20, i30, h33)))
+ }
+
+ test("countIncomplete-2") {
+ assert(0 === countIncompleteApplications(Nil))
+ }
+
+ test("countIncomplete-3") {
+ assert(0 === countIncompleteApplications(List(h44, h33)))
+ }
+
+ test("findStartOfWindow") {
+ assert(Some(i20) === findStartOfWindow(List(h44, i20, i30, h33)))
+ }
+
+ test("findStartOfWindow-2") {
+ assert(Some(h44) === findStartOfWindow(List(h44, h12, h33)))
+ }
+
+ test("combineResults-list-nil") {
+ assert((h44 :: Nil) === combineResults(List(h44), Nil))
+ }
+
+ test("combineResults-2-Nil-list") {
+ assert((h44 :: Nil) === combineResults(Nil, List(h44)))
+ }
+
+ test("combineResults-3-Nil-lists") {
+ assert(Nil === combineResults(Nil, Nil))
+ }
+
+ test("combineResults-5") {
+ assert((h44 :: i20 :: Nil) === combineResults(List(h44), List(i20)))
+ }
+
+ test("combineResults-6-merge-duplicate-to-one") {
+ assert(List(h44) === combineResults(List(h44), List(h44)))
+ }
+
+ test("combineResults-7-completed") {
+ assert(List(iA11_completed) === combineResults(List(iA10_incomplete), List(iA11_completed)))
+ }
+
+ test("merge-multiple_attempts") {
+ assert(List(historyInfo(a1_attempt_1, a1_attempt_2.attempts ++ a1_attempt_1.attempts))
+ === combineResults(List(a1_attempt_1), List(a1_attempt_2)))
+ }
+
+ test("SortApplications-1") {
+ assert((h33 :: h44 :: Nil) === sortApplicationsByStartTime(List(h44, h33)))
+ }
+
+ test("SortApplications-2") {
+ assert((h22 :: i20 :: h33 :: Nil) === sortApplicationsByStartTime(List(h22, i20, h33)))
+ }
+
+ test("SortApplications-3") {
+ assert((i20 :: h22 :: Nil) === sortApplicationsByStartTime(List(i20, h22)))
+ }
+
+ test("findLatest") {
+ assert(Some(h22) === findLatestApplication(List(h12, h22, i20)))
+ }
+
+ test("findLatest-2") {
+ assert(Some(h22) === findLatestApplication(List(h22, i20)))
+ }
+
+ test("findLatest-3") {
+ assert(Some(i20) === findLatestApplication(List(h12, i20)))
+ }
+
+ test("buildEntityIds") {
+ val sparkAppId = "spark_app_id_2"
+ val attempt = "attempt_id"
+ val yarnAppStr = yarnAppId.toString
+ val attemptId = Some(yarnAttemptId)
+ val attemptIdStr = yarnAttemptId.toString
+ assert(attemptIdStr === buildEntityId(yarnAppId, attemptId),
+ "all fields")
+ assert(attemptIdStr === buildEntityId(yarnAppId, attemptId),
+ "no attempt ID")
+ assert(yarnAppStr === buildEntityId(yarnAppId, None), "yarnAppId only")
+ }
+
+ test("buildApplicationAttemptIdField") {
+ val sparkAppId = "spark_app_id_2"
+ val attempt = "attempt_id"
+ assert(attempt === buildApplicationAttemptIdField(Some(attempt)),
+ "all fields")
+ assert(SINGLE_ATTEMPT === buildApplicationAttemptIdField(None),
+ "attempt = None")
+ }
+
+ test("EntityAndBack") {
+ val sparkAppId = Some("spark-app-id-1")
+ val yarnAppStr = yarnAppId.toString
+ val sparkAttemptId = Some("spark-attempt-id")
+ val yarnAttemptIdStr = yarnAttemptId.toString
+
+ val entity = createTimelineEntity(yarnAppId,
+ Some(yarnAttemptId),
+ sparkAppId,
+ sparkAttemptId,
+ "app",
+ "user",
+ 1000, 0, 1000)
+ val entityDescription = describeEntity(entity)
+ val ev1 = entity.getOtherInfo.get(FIELD_ENTITY_VERSION)
+ val version = numberField(entity, FIELD_ENTITY_VERSION, -1).longValue()
+ assert (0 < version, s"wrong version in $entityDescription")
+
+ // build an TimelineApplicationHistoryInfo instance
+ val info = toApplicationHistoryInfo(entity)
+ assert(yarnAppStr === info.id, "info.id")
+ val attempt = info.attempts.head
+ assert(sparkAttemptId === attempt.attemptId, s"attempt.attemptId in $attempt")
+ assert(yarnAttemptIdStr === attempt.entityId, s"attempt.entityId in $attempt")
+ assert(version === attempt.version, s"version in $attempt")
+ }
+
+ test("EntityWithoutAttempt") {
+ val sparkAppId = Some("spark-app-id-1")
+ val yarnAppStr = yarnAppId.toString
+ val yarnAttemptIdStr = yarnAttemptId.toString
+
+ val entity = createTimelineEntity(yarnAppId,
+ None,
+ sparkAppId,
+ None,
+ "app",
+ "user",
+ 1000, 0, 1000)
+ val info = toApplicationHistoryInfo(entity)
+ assert(yarnAppStr === info.id)
+
+ val attempt = info.attempts.head
+ assert("1" === attempt.attemptId.get, "attempt.attemptId")
+ assert(yarnAppStr === attempt.entityId, "attempt.entityId")
+ }
+
+ test("MergeHistoryEvents") {
+ val one_1 = new TimelineApplicationHistoryInfo("app1", "one", attempt_1_1 :: Nil)
+ val one_2 = new TimelineApplicationHistoryInfo("app1", "one", attempt_1_2 :: Nil)
+ val one_0 = new TimelineApplicationHistoryInfo("app1", "one", Nil)
+
+ val merge_12 = mergeAttempts(one_1, one_2)
+ assertListSize( merge_12.attempts, 2, "merged attempt list")
+ assert(List(attempt_1_2, attempt_1_1) === merge_12.attempts)
+
+ val merge_10 = mergeAttempts(one_1, one_0)
+ assert(1 === merge_10.attempts.size)
+ assert(one_1 === mergeAttempts(one_1, one_1))
+ }
+
+ test("MergeHistoryEventsIdNone") {
+ val one_1 = new TimelineApplicationHistoryInfo("1", "one", none_incomplete :: Nil)
+ val one_2 = new TimelineApplicationHistoryInfo("1", "one", none_completed :: Nil)
+ val one_0 = new TimelineApplicationHistoryInfo("1", "one", Nil)
+ }
+
+ test("merge-results-None-attemptId-incomplete-first") {
+ assert(List(none_completed) === mergeAttemptInfoLists(List(none_incomplete),
+ List(none_completed)))
+ }
+
+ test("merge-results-None-attemptId-incomplete-second") {
+ // and in the other order
+ assert(List(none_completed) === mergeAttemptInfoLists(List(none_completed),
+ List(none_incomplete)))
+ }
+
+ test("MergeAttemptOrdering-1") {
+ assert(none_completed === mostRecentAttempt(none_completed, none_incomplete))
+ }
+
+ test("MergeAttemptOrdering-2") {
+ assert(none_completed === mostRecentAttempt(none_incomplete, none_completed))
+ }
+
+ test("MergeAttemptOrdering-3") {
+ assert(none_completed === mostRecentAttempt(none_incomplete, none_completed))
+ }
+
+ test("MergeAttemptOrdering-4") {
+ assert(attempt_1_1_updated === mostRecentAttempt(attempt_1_1, attempt_1_1_updated))
+ }
+
+ test("MergeAttemptOrdering-5") {
+ assert(attempt_1_1_updated === mostRecentAttempt(attempt_1_1_updated, attempt_1_1))
+ }
+
+ test("MergeAttemptOrdering-6") {
+ assert(none_completed_orig_time ===
+ mostRecentAttempt(none_incomplete, none_completed_orig_time))
+ }
+
+ test("MergeAttemptOrdering-7") {
+ assert(none_completed_orig_time ===
+ mostRecentAttempt(none_completed_orig_time, none_incomplete))
+ }
+
+ test("MergeAttemptOrdering-8") {
+ assert(attempt_1_1_updated_version === mostRecentAttempt(attempt_1_1,
+ attempt_1_1_updated_version))
+ }
+
+ test("MergeAttemptOrdering-9") {
+ assert(attempt_1_1_updated_version === mostRecentAttempt(attempt_1_1_updated_version,
+ attempt_1_1))
+ }
+
+ test("MergeAttemptOrdering-10") {
+ assert(attempt_1_1_completed_v4 === mostRecentAttempt(attempt_1_1_updated_version,
+ attempt_1_1_completed_v4))
+ }
+
+ test("MergeAttemptOrdering-11") {
+ assert(attempt_1_1_completed_v4 === mostRecentAttempt(attempt_1_1_completed_v4,
+ attempt_1_1_updated_version))
+ }
+
+ test("SortAttempts-by-version") {
+ assert((i20 :: h22 :: Nil) === sortApplicationsByStartTime(List(i20, h22)))
+ }
+
+ test("merge-results-updated-first") {
+ // and in the other order
+ assert(List(attempt_1_1_updated_version) ===
+ mergeAttemptInfoLists(List(attempt_1_1_updated_version), List(attempt_1_1)))
+ }
+
+ test("merge-results-updated-second") {
+ // and in the other order
+ assert(List(attempt_1_1_updated_version) ===
+ mergeAttemptInfoLists(List(attempt_1_1), List(attempt_1_1_updated_version)))
+ }
+
+ test("merge-results-v2_v3") {
+ // and in the other order
+ assert(List(attempt_1_2, attempt_1_1_updated_version) ===
+ mergeAttemptInfoLists(List(attempt_1_2), List(attempt_1_1_updated_version)))
+ }
+
+ test("merge-results-v1_v2_v3") {
+ // and in the other order
+ assert(List(attempt_1_2, attempt_1_1_updated_version) ===
+ mergeAttemptInfoLists(List(attempt_1_2, attempt_1_1),
+ List(attempt_1_1_updated_version)))
+ }
+
+
+ test("merge-multiple-attempts-by-version") {
+
+ val app2 = new TimelineApplicationHistoryInfo("app2", "app2", List(attempt_1_1))
+
+ val histories1 = combineResults(Nil, List(app2))
+ val app2_updated = new TimelineApplicationHistoryInfo("app2", "app2",
+ List(attempt_1_1_completed_v4))
+
+ val histories2 = combineResults(Nil, List(app2_updated))
+ val merged = combineResults(histories1, histories2)
+ merged should have size 1
+ val finalApp = merged.head
+ val finalAppDescription = describeApplicationHistoryInfo(finalApp)
+ val finalAttempt = finalApp.attempts.head
+ assert(finalAttempt.completed, s"not completed $finalAppDescription")
+ assert(List(historyInfo(a1_attempt_1, a1_attempt_2.attempts ++ a1_attempt_1.attempts))
+ === combineResults(List(a1_attempt_1), List(a1_attempt_2)))
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/publish/ContextEventPublishingSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/publish/ContextEventPublishingSuite.scala
new file mode 100644
index 000000000000..4cdf4085b2ce
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/publish/ContextEventPublishingSuite.scala
@@ -0,0 +1,67 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.publish
+
+import java.io.File
+
+import org.apache.spark.deploy.history.yarn.integration.AbstractHistoryIntegrationTests
+import org.apache.spark.deploy.history.yarn.testtools.{HistoryServiceListeningToSparkContext, TimelineSingleEntryBatchSize}
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+
+/**
+ * hookup from spark context to timeline.
+ */
+class ContextEventPublishingSuite
+ extends AbstractHistoryIntegrationTests
+ with HistoryServiceListeningToSparkContext
+ with TimelineSingleEntryBatchSize {
+
+ test ("Publish events via Context") {
+ describe("Publish events via Context")
+ // hook up to spark context
+ historyService = startHistoryService(sc)
+ assert(historyService.listening, s"listening $historyService")
+ assert(1 === historyService.batchSize, s"batch size in $historyService")
+ assert(historyService.bondedToATS, s"not bonded to ATS: $historyService")
+ // post in an app start
+ var flushes = 0
+ logDebug("posting app start")
+ enqueue(appStartEvent())
+ flushes += 1
+ awaitEmptyQueue(historyService, TEST_STARTUP_DELAY)
+
+ // add a local file to the context, so generate another event
+ val tempFile = File.createTempFile("test", ".txt")
+ sc.addFile(tempFile.toURI.toString)
+ tempFile.delete()
+ flushes += 1
+
+ // closing context generates an application stop
+ logDebug("stopping context")
+ sc.stop()
+ awaitEmptyQueue(historyService, TEST_STARTUP_DELAY)
+
+ flushes += 1
+ logDebug(s"status: $historyService")
+
+ // seeing intermittent failure
+ awaitFlushCount(historyService, flushes, TEST_STARTUP_DELAY)
+ flushHistoryServiceToSuccess()
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/publish/TimelinePostSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/publish/TimelinePostSuite.scala
new file mode 100644
index 000000000000..a9e6c4e1d6bc
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/publish/TimelinePostSuite.scala
@@ -0,0 +1,40 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.publish
+
+import org.apache.spark.deploy.history.yarn.YarnEventListener
+import org.apache.spark.deploy.history.yarn.integration.AbstractHistoryIntegrationTests
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+
+class TimelinePostSuite extends AbstractHistoryIntegrationTests {
+
+ test("Round Trip App Stop") {
+ historyService = startHistoryService(sc)
+ val outcome = postEvent(appStartEvent(), now())
+ historyService.stop()
+ awaitEmptyQueue(historyService, TEST_STARTUP_DELAY)
+ }
+
+ test("App Start Via Event Listener") {
+ historyService = startHistoryService(sc)
+ val listener = new YarnEventListener(sc, historyService)
+ val sparkEvt = appStartEvent()
+ listener.onApplicationStart(sparkEvt)
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/AbstractYarnHistoryTests.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/AbstractYarnHistoryTests.scala
new file mode 100644
index 000000000000..d0fc8048c07e
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/AbstractYarnHistoryTests.scala
@@ -0,0 +1,133 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.testtools
+
+import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
+import org.scalatest.{BeforeAndAfter, Matchers}
+
+import org.apache.spark.{LocalSparkContext, Logging, SparkConf, SparkContext, SparkFunSuite}
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+import org.apache.spark.deploy.history.yarn.{YarnEventListener, YarnHistoryService}
+import org.apache.spark.scheduler.cluster.SchedulerExtensionServiceBinding
+
+/**
+ * This is a base class for the YARN history test suites, both mock and integration
+ *
+ * Subclasses are expected to use traits and/or overriding of
+ * [[ContextSetup.setupConfiguration()]]
+ * to tune the configuration of the instantiated context.
+ *
+ * To avoid any ambiguity about the ordering/invocation of
+ * any before/after code, the operations are passed to
+ * overriddeable `setup()` and `teardown()`
+ * invocations. Subclasses must relay the method calls to their
+ * superclasses to ensure correct setup and teardown.
+ */
+abstract class AbstractYarnHistoryTests
+ extends SparkFunSuite
+ with TimelineOptionsInContext
+ with TimelineServiceDisabled
+ with HistoryServiceNotListeningToSparkContext
+ with LocalSparkContext
+ with BeforeAndAfter
+ with Logging
+ with ExtraAssertions
+ with Matchers {
+
+ /*
+ * Setup phase creates the spark context, and anything else which tests require
+ */
+ before {
+ setup()
+ }
+
+ /*
+ * Setup creates the spark context
+ */
+ protected def setup(): Unit = {
+ val sparkConf = new SparkConf()
+
+ if (sc != null) {
+ fail("Spark Context is not null -a previous test did not clean up properly")
+ }
+
+ sc = createSparkContext(sparkConf)
+ logDebug(s"Created context $sc")
+ }
+
+ /**
+ * Create the spark context
+ * @param sparkConf configuration to extend
+ */
+ protected def createSparkContext(sparkConf: SparkConf): SparkContext = {
+ sparkConf.setMaster("local").setAppName("AbstractYarnHistoryTests")
+ logInfo("Creating a new spark context")
+ new SparkContext(setupConfiguration(sparkConf))
+ }
+
+ /**
+ * Create and start a history service.
+ *
+ * @param sc context
+ * @param id application ID
+ * @param appAttemptId optional attempt ID
+ * @return the instantiated service
+ */
+ protected def startHistoryService(
+ sc: SparkContext,
+ id: ApplicationId = applicationId,
+ appAttemptId: Option[ApplicationAttemptId] = Some(attemptId)): YarnHistoryService = {
+ assertNotNull(sc, "Spark context")
+ val service = new YarnHistoryService()
+ service.start(SchedulerExtensionServiceBinding(sc, id, appAttemptId))
+ assert(YarnHistoryService.StartedState === service.serviceState, s"wrong state: $service")
+ service
+ }
+
+ /**
+ * Create a history service, post an event sequence, then stop the service.
+ * The @ of attempts posted is returned in the response to stop tests being brittle
+ * against the numbe of events posted automatically. This is the code which posts the events
+ * -it's the code that can reliably announce this.
+ * @param sc context
+ * @return (the (now closed) history service, the count of events posted for use in assertions)
+ */
+ def postEvents(sc: SparkContext): (YarnHistoryService, Int) = {
+ val service: YarnHistoryService = startHistoryService(sc)
+ val listener = new YarnEventListener(sc, service)
+ var eventsPosted = 0
+ try {
+ listener.onApplicationStart(applicationStart)
+ listener.onEnvironmentUpdate(environmentUpdate)
+ listener.onApplicationEnd(applicationEnd)
+ eventsPosted += 3
+ // wait for them all to be processed
+ awaitEventsProcessed(service, eventsPosted, TEST_STARTUP_DELAY)
+ // only 2 post events, as the update does not trigger a post unless batch size == 1
+ awaitAtLeast(2, TEST_STARTUP_DELAY,
+ () => service.postAttempts,
+ () => s"Post count in $service")
+ } finally {
+ // because the events have been processed and waited for,
+ // a duplicate end event will not be posted
+ service.stop()
+ }
+ (service, eventsPosted)
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/ContextSetup.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/ContextSetup.scala
new file mode 100644
index 000000000000..cbfc208b7e77
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/ContextSetup.scala
@@ -0,0 +1,91 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.testtools
+
+import org.apache.hadoop.yarn.conf.YarnConfiguration
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.history.yarn.YarnHistoryService
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+
+/**
+ * Trait implemented by everything setting up a context; the model is that
+ * the traits can be chained, with the final state determined by the order
+ *
+ * 1. base implementation does nothing.
+ * 2. subclass traits are expected to call the superclass first, then
+ * apply their own options.
+ */
+trait ContextSetup {
+
+ def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ sparkConf
+ }
+}
+
+trait TimelineServiceEnabled extends ContextSetup {
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ hadoopOpt(super.setupConfiguration(sparkConf),
+ YarnConfiguration.TIMELINE_SERVICE_ENABLED, "true")
+ }
+}
+
+trait TimelineServiceDisabled extends ContextSetup {
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ hadoopOpt(super.setupConfiguration(sparkConf),
+ YarnConfiguration.TIMELINE_SERVICE_ENABLED, "false")
+ }
+}
+
+/**
+ * Add the timeline options
+ */
+trait TimelineOptionsInContext extends ContextSetup {
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ YarnTestUtils.addBasicTimelineOptions(super.setupConfiguration(sparkConf))
+ }
+}
+
+/**
+ * request that created history services register with the spark context for lifecycle events
+ */
+trait HistoryServiceListeningToSparkContext extends ContextSetup {
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ super.setupConfiguration(sparkConf).set(YarnHistoryService.REGISTER_LISTENER, "true")
+ }
+}
+
+/**
+ * request that created history services are not registered with the spark context for
+ * lifecycle events
+ */
+trait HistoryServiceNotListeningToSparkContext extends ContextSetup {
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ super.setupConfiguration(sparkConf).set(YarnHistoryService.REGISTER_LISTENER, "false")
+ }
+}
+
+/**
+ * Switch to single entry batch sizes
+ */
+trait TimelineSingleEntryBatchSize extends ContextSetup {
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ super.setupConfiguration(sparkConf).set(YarnHistoryService.BATCH_SIZE, "1")
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/ExtraAssertions.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/ExtraAssertions.scala
new file mode 100644
index 000000000000..e012694a0733
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/ExtraAssertions.scala
@@ -0,0 +1,247 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.testtools
+
+import java.util.{Collection => JCollection}
+
+import org.apache.hadoop.service.Service
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity
+import org.scalatest.Assertions
+
+import org.apache.spark.Logging
+import org.apache.spark.deploy.history.yarn.YarnHistoryService
+import org.apache.spark.deploy.history.yarn.YarnTimelineUtils._
+
+/**
+ * Miscellaneous extra assertions
+ */
+trait ExtraAssertions extends Logging with Assertions {
+
+ /**
+ * Assert that an exception's toString value contains the supplied text.
+ * If not, an error is logged and the exception is rethrown
+ *
+ * @param ex exception
+ * @param text text
+ */
+ def assertExceptionMessageContains(ex: Exception, text: String): Unit = {
+ if (!ex.toString.contains(text)) {
+ logError(s"Did not find text $text in $ex", ex)
+ throw ex
+ }
+ }
+
+ /**
+ * Check the exception message text and toString check
+ *
+ * @param ex exception to examine
+ * @param messageCheck string to check in the `Exception.getMessage()` string
+ * @param stringCheck string to look for in the `Exception.toString()` string;
+ * if empty the check is skipped.
+ */
+ def assertExceptionDetails(ex: Throwable, messageCheck: String, stringCheck: String): Unit = {
+ assertNotNull(ex.getMessage, s"Exception message in $ex")
+ if (!ex.getMessage.contains(messageCheck)) {
+ throw ex
+ }
+ if (!stringCheck.isEmpty && !ex.toString.contains(stringCheck)) {
+ throw ex
+ }
+ }
+
+ /**
+ * Assert that a value is not null
+ *
+ * @param value value
+ * @param text text for assertion
+ */
+ def assertNotNull(value: Any, text: String): Unit = {
+ assert(value !== null, s"Null value; $text")
+ }
+
+ /**
+ * Assert that an optional value is not `None`.
+ *
+ * @param value value
+ * @param text text for assertion
+ */
+ def assertSome(value: Option[Any], text: String): Unit = {
+ assert(value.nonEmpty, s"optional value is None; $text")
+ }
+
+ /**
+ * Assert that an optional value is `None`
+ *
+ * @param value value
+ * @param text text for assertion
+ */
+ def assertNone(value: Option[Any], text: String): Unit = {
+ assert(value.isEmpty, s"Optional value is $value.get; $text")
+ }
+
+ /**
+ * Assert that a Spark traversable instance is not empty
+ *
+ * @param traversable the traversable to check
+ * @param text text for assertion
+ * @tparam T traversable type
+ */
+ def assertNotEmpty[T](traversable: Traversable[T], text: String): Unit = {
+ assert(traversable.nonEmpty, s"Empty traversable; $text")
+ }
+
+ /**
+ * Assert that a java collection is not empty
+ *
+ * @param collection the collection to check
+ * @param text text for assertion
+ * @tparam T collection type
+ */
+ def assertNotEmpty[T](collection: JCollection[T], text: String): Unit = {
+ assert (!collection.isEmpty, s"Empty collection; $text")
+ }
+
+ /**
+ * Assert the list is of the given size. if not all elements are logged @ error,
+ * then the method raises a failure.
+ *
+ * @param list list to examine
+ * @param expectedSize expected size
+ * @param message error message
+ * @tparam T list type
+ */
+ def assertListSize[T](list: Seq[T], expectedSize: Int, message: String): Unit = {
+ assertNotNull(list, message)
+ if (list.size != expectedSize) {
+ // no match
+ val errorText = s"Wrong list size: expected=$expectedSize actual=${list.size}: $message"
+ logError(errorText)
+ list.foreach { e => logError(e.toString) }
+ fail(errorText)
+ }
+ }
+
+ /**
+ * Assert that a list is Nil (and implicitly, not null)
+ *
+ * If not, an assertion is raised that contains the message and the list
+ * @param list list to check
+ * @param message message to raise
+ * @tparam T list type
+ */
+ def assertNil[T](list: Seq[T], message: String): Unit = {
+ assertNotNull(list, message)
+ if (list != Nil) {
+ fail(message + " " + list)
+ }
+ }
+
+ /**
+ * Assert that a service is not listening
+ *
+ * @param historyService history service
+ */
+ def assertNotListening(historyService: YarnHistoryService): Unit = {
+ assert(!historyService.listening, s"history service is listening for events: $historyService")
+ }
+
+ /**
+ * Assert that the number of events processed matches the number expected
+ *
+ * @param historyService history service
+ * @param expected expected number
+ * @param details text to include in error messages
+ */
+ def assertEventsProcessed(historyService: YarnHistoryService,
+ expected: Int, details: String): Unit = {
+ assertResult(expected, "wrong number of events processed " + details) {
+ historyService.eventsProcessed
+ }
+ }
+
+ /**
+ * Assert that two timeline entities are non-null and equal
+ *
+ * @param expected expected entry
+ * @param actual actual
+ */
+ def assertEntitiesEqual(expected: TimelineEntity, actual: TimelineEntity): Unit = {
+ require(expected != null)
+ require(actual != null)
+ assert(expected === actual,
+ s"Expected ${describeEntity(expected)}; got ${describeEntity(actual)}}")
+ }
+
+ /**
+ * Assert that a service is in a specific state
+ *
+ * @param service service
+ * @param state required state
+ */
+ def assertInState(service: Service, state: Service.STATE): Unit = {
+ assertNotNull(service, "null service")
+ assert(service.isInState(state), s"not in state $state: $service")
+ }
+
+ /**
+ * Assert that a source string contains the `contained` substring.
+ * (This is not a check for a proper subset; equality is also acceptable)
+ * @param source source string
+ * @param contained string to look for
+ */
+ def assertContains(source: String, contained: String, text: String = ""): Unit = {
+ assertNotNull(source, s"$text null source")
+ assertNotNull(contained, s"$text null `contained`")
+ if (!source.contains(contained)) {
+ fail(s"$text -Did not find '$contained' in '$source'")
+ }
+ }
+
+ /**
+ * Assert that a source string does contains the `contained` substring.
+ * @param source source string
+ * @param contained string to look for
+ */
+ def assertDoesNotContain(source: String, contained: String, text: String = ""): Unit = {
+ assertNotNull(source, s"$text null source")
+ assertNotNull(contained, s"$text null `contained`")
+ if (source.contains(contained)) {
+ fail(s"$text -Found '$contained' in '$source'")
+ }
+ }
+
+ /**
+ * Assert that a `[String, String]` map contains a `key:value` mapping,
+ * and that the value contains the specified text.
+ * @param map map to query
+ * @param key key to retrieve
+ * @param text text to look for in the resolved value
+ */
+ protected def assertMapValueContains(map: Map[String, String],
+ key: String, text: String): Unit = {
+ map.get(key) match {
+ case Some(s) =>
+ if (!text.isEmpty && !s.contains(text)) {
+ fail(s"Did not find '$text' in key[$key] = '$s'")
+ }
+
+ case None =>
+ fail(s"No entry for key $key")
+ }
+ }
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/FreePortFinder.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/FreePortFinder.scala
new file mode 100644
index 000000000000..7174f4bbcbc5
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/FreePortFinder.scala
@@ -0,0 +1,101 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.testtools
+
+import java.net.{InetAddress, ServerSocket}
+
+import scala.collection.mutable.ListBuffer
+
+/**
+ * Trait to find free ports on localhost
+ */
+trait FreePortFinder {
+
+ /**
+ * Find a free port by listening on port 0
+ * @return
+ */
+ def findPort(): Int = {
+ tryToListen(0)._2
+ }
+
+ /**
+ * Simple function to see if a port is free; if it is return the address and the port allocated.
+ *
+ * This function can be passed to `Util.startServiceOnPort`
+ * @param port port to try. If 0, the OS chooses the port
+ * @return an (address, port) tuple
+ */
+ def tryToListen(port: Int): (InetAddress, Int) = {
+ val socket = new ServerSocket(port)
+ val address = socket.getInetAddress
+ val localPort = socket.getLocalPort
+ socket.close()
+ (address, localPort)
+ }
+
+ /**
+ * Return the value of the local host address -defaults to 127.0.0.1
+ * @return the address to use for local/loopback addresses.
+ */
+ def localIPv4Address(): String = {
+ "127.0.0.1"
+ }
+
+ /**
+ * Get a local address as an address:port string and an integer port value
+ * @return a free port to bind to
+ */
+ def findIPv4AddressAsPortPair(): (String, Int) = {
+ val port = findPort()
+ (localhostAndPort(port), port)
+ }
+
+ /**
+ * Given a port, return a localhost:port pair
+ * @param port port
+ * @return the name for the localhost for test runs.
+ */
+ def localhostAndPort(port: Int): String = {
+ localIPv4Address() + ":" + port
+ }
+
+ /**
+ * Find the specified number of unique ports. This is done in parallel, so the
+ * ports are guaranteed to be different. The ports are all closed afterwards,
+ * so other network services started may grab those same ports.
+ *
+ * @param count number of ports to find.
+ * @return a list of ports to use.
+ */
+ def findUniquePorts(count: Integer): Seq[Integer] = {
+ val sockets = new ListBuffer[ServerSocket]()
+ val ports = new ListBuffer[Integer]()
+ for (i <- 1 to count) {
+ val socket = new ServerSocket(0)
+ sockets += socket
+ ports += socket.getLocalPort
+ }
+ sockets.foreach(_.close())
+ // Uniqueness: foreach port, there only exists one port in the list which has that value
+ require(ports.map( p => ports.count(_ == p) == 1).count(_ == true) == count,
+ s"Duplicate port allocation in " + ports.mkString(","))
+
+ ports
+ }
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/YarnTestUtils.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/YarnTestUtils.scala
new file mode 100644
index 000000000000..8ee3c3a0bf5f
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/testtools/YarnTestUtils.scala
@@ -0,0 +1,594 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.testtools
+
+import java.io.IOException
+import java.net.URL
+
+import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity
+import org.apache.hadoop.yarn.conf.YarnConfiguration
+import org.apache.hadoop.yarn.server.timeline.MemoryTimelineStore
+
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.deploy.history.yarn.YarnHistoryService._
+import org.apache.spark.deploy.history.yarn.{YarnHistoryService, YarnTimelineUtils}
+import org.apache.spark.scheduler.cluster.{StubApplicationAttemptId, StubApplicationId}
+import org.apache.spark.scheduler.{JobFailed, JobSucceeded, SparkListenerApplicationEnd, SparkListenerApplicationStart, SparkListenerEnvironmentUpdate, SparkListenerEvent, SparkListenerJobEnd, SparkListenerJobStart}
+import org.apache.spark.util.Utils
+
+object YarnTestUtils extends ExtraAssertions with FreePortFinder {
+
+ val environmentUpdate = SparkListenerEnvironmentUpdate(Map[String, Seq[(String, String)]](
+ "JVM Information" -> Seq(("GC speed", "9999 objects/s"), ("Java home", "Land of coffee")),
+ "Spark Properties" -> Seq(("Job throughput", "80000 jobs/s, regardless of job type")),
+ "System Properties" -> Seq(("Username", "guest"), ("Password", "guest")),
+ "Classpath Entries" -> Seq(("Super library", "/tmp/super_library"))))
+
+ /**
+ * Application name used in the app start event and tests
+ */
+ val APP_NAME = "spark-demo"
+
+ /**
+ * User submitting the job
+ */
+ val APP_USER = "data-scientist"
+
+ /**
+ * application ID
+ */
+ val APP_ID = "application_id_0001"
+
+ /**
+ * Spark option to set for the history provider
+ */
+ val SPARK_HISTORY_PROVIDER = "spark.history.provider"
+
+ /**
+ * Constant used to define history port in Spark `HistoryServer` class
+ */
+ val SPARK_HISTORY_UI_PORT = "spark.history.ui.port"
+
+ val completedJobsMarker = "Completed Jobs (1)"
+ val activeJobsMarker = "Active Jobs (1)"
+
+
+ /**
+ * Time to wait for anything to start/state to be reached
+ */
+ val TEST_STARTUP_DELAY = 5000
+
+ /** probes during service shutdown need to handle delayed posting */
+ val SERVICE_SHUTDOWN_DELAY = 10000
+
+ /**
+ * Cancel a test if the network isn't there.
+ *
+ * If called during setup, this will abort the test
+ */
+ def cancelIfOffline(): Unit = {
+
+ try {
+ val hostname = Utils.localHostName()
+ log.debug(s"local hostname is $hostname")
+ } catch {
+ case ex: IOException =>
+ cancel(s"Localhost name not known: $ex", ex)
+ }
+ }
+
+ /**
+ * Return a time value
+ *
+ * @return the current time in milliseconds
+ */
+ def now(): Long = {
+ System.currentTimeMillis()
+ }
+
+ /**
+ * Get a time in the future
+ *
+ * @param millis future time in millis
+ * @return now + the time offset
+ */
+ def future(millis: Long): Long = {
+ now() + millis
+ }
+
+ /**
+ * Log an entry with a line either side. This aids splitting up tests from the noisy logs
+ *
+ * @param text text to log
+ */
+ def describe(text: String): Unit = {
+ logInfo(s"\nTest:\n $text\n\n")
+ }
+
+ /**
+ * Set a hadoop opt in the config.
+ *
+ * This adds the `"spark.hadoop."` prefix to all entries which do not already have it
+ *
+ * @param sparkConfig target configuration
+ * @param key hadoop option key
+ * @param value value
+ */
+ def hadoopOpt(sparkConfig: SparkConf, key: String, value: String): SparkConf = {
+ if (key.startsWith("spark.hadoop.")) {
+ sparkConfig.set(key, value)
+ } else {
+ sparkConfig.set("spark.hadoop." + key, value)
+ }
+ }
+
+ /**
+ * Bulk set of an entire map of Hadoop options
+ *
+ * @param sparkConfig target configuration
+ * @param options option map
+ */
+ def applyHadoopOptions(sparkConfig: SparkConf, options: Map[String, String]): SparkConf = {
+ options.foreach( e => hadoopOpt(sparkConfig, e._1, e._2))
+ sparkConfig
+ }
+
+ /**
+ * Apply the basic timeline options to the hadoop config
+ *
+ * @return the modified config
+ */
+ def addBasicTimelineOptions(sparkConf: SparkConf): SparkConf = {
+ val ports = findUniquePorts(3)
+ applyHadoopOptions(sparkConf,
+ Map(YarnConfiguration.TIMELINE_SERVICE_ENABLED -> "true",
+ YarnConfiguration.TIMELINE_SERVICE_ADDRESS -> localhostAndPort(ports.head),
+ YarnConfiguration.TIMELINE_SERVICE_WEBAPP_ADDRESS -> localhostAndPort(ports(1)),
+ YarnConfiguration.TIMELINE_SERVICE_STORE -> classOf[MemoryTimelineStore].getName,
+ YarnConfiguration.TIMELINE_SERVICE_CLIENT_MAX_RETRIES -> "1",
+ YarnConfiguration.TIMELINE_SERVICE_CLIENT_RETRY_INTERVAL_MS -> "200"))
+
+ // final port in the set
+ sparkConf.set(SPARK_HISTORY_UI_PORT, ports(2).toString)
+
+ // turn off the minimum refresh interval.
+ // uses a string to it can be set from code that doesn't refer to any provider-side
+ // classes
+ sparkConf.set("spark.history.yarn.min-refresh-interval", "0")
+
+ // shorter reset interval and shutdown time
+ sparkConf.set(POST_RETRY_INTERVAL, "10ms")
+ sparkConf.set(SHUTDOWN_WAIT_TIME, "5s")
+ }
+
+ /**
+ * Convert the single timeline event in a timeline entity to a spark event
+ *
+ * @param entity entity to convert, which must contain exactly one event.
+ * @return the converted event
+ */
+ def convertToSparkEvent(entity: TimelineEntity): SparkListenerEvent = {
+ assertResult(1, "-wrong # of events in the timeline entry") {
+ entity.getEvents().size()
+ }
+ YarnTimelineUtils.toSparkEvent(entity.getEvents().get(0))
+ }
+
+ /**
+ * Create an app start event, using the fixed [[APP_NAME]] and [[APP_USER]] values
+ * for appname and user; no attempt ID
+ *
+ * @param time application start time
+ * @param appId event ID; default is [[APP_ID]]
+ * @return the event
+ */
+ def appStartEventWithAttempt(time: Long = 1,
+ appId: String,
+ user: String,
+ attemptId: ApplicationAttemptId): SparkListenerApplicationStart = {
+ appStartEvent(time, appId, user, Some(attemptId.toString))
+ }
+
+ /**
+ * Create an app start event
+ *
+ * @param time application start time
+ * @param appId event ID; default is [[APP_ID]]
+ * @param user the user; defaults is [[APP_USER]]
+ * @param attempt attempt ID; default is `None`
+ * @return the event
+ */
+ def appStartEvent(time: Long = 1434920400000L,
+ appId: String = APP_ID,
+ user: String = APP_USER,
+ attempt: Option[String] = None,
+ name: String = APP_NAME): SparkListenerApplicationStart = {
+ require(name != null)
+ require(appId != null)
+ require(user != null)
+ SparkListenerApplicationStart(name, Some(appId), time, user, attempt)
+ }
+
+ def appStartEvent(time: Long, ctx: SparkContext): SparkListenerApplicationStart = {
+ appStartEvent(time, ctx.applicationId, ctx.sparkUser, ctx.applicationAttemptId)
+ }
+
+ def appStopEvent(time: Long = 1): SparkListenerApplicationEnd = {
+ new SparkListenerApplicationEnd(time)
+ }
+
+ def jobStartEvent(time: Long, id: Int) : SparkListenerJobStart = {
+ new SparkListenerJobStart(id, time, Nil, null)
+ }
+
+ def jobSuccessEvent(time: Long, id: Int) : SparkListenerJobEnd = {
+ new SparkListenerJobEnd(id, time, JobSucceeded)
+ }
+
+ def jobFailureEvent(time: Long, id: Int, ex: Exception) : SparkListenerJobEnd = {
+ new SparkListenerJobEnd(id, time, JobFailed(ex))
+ }
+
+ def newEntity(time: Long): TimelineEntity = {
+ val entity = new TimelineEntity
+ entity.setStartTime(time)
+ entity.setEntityId("post")
+ entity.setEntityType(YarnHistoryService.SPARK_EVENT_ENTITY_TYPE)
+ entity
+ }
+
+ val applicationId: ApplicationId = new StubApplicationId(0, 1111L)
+ val attemptId: ApplicationAttemptId = new StubApplicationAttemptId(applicationId, 1)
+ val applicationStart = appStartEventWithAttempt(now(), applicationId.toString, "bob", attemptId)
+ val applicationEnd = SparkListenerApplicationEnd(now() + 60000)
+
+ /**
+ * Outcomes of probes
+ */
+ sealed abstract class Outcome
+ case class Fail() extends Outcome
+ case class Retry() extends Outcome
+ case class Success() extends Outcome
+ case class TimedOut() extends Outcome
+
+ /**
+ * Spin and sleep awaiting an observable state.
+ *
+ * The scalatest `eventually` operator is similar, and even adds exponential backoff.
+ * What this offers is:
+ *
+ * 1. The ability of the probe to offer more than just success/fail, but a "fail fast"
+ * operation which stops spinning early.
+ * 2. A detailed operation to invoke on failure, so provide more diagnostics than
+ * just the assertion.
+ *
+ * @param interval sleep interval
+ * @param timeout time to wait
+ * @param probe probe to execute
+ * @param failure closure invoked on timeout/probe failure
+ */
+ def spinForState(description: String,
+ interval: Long,
+ timeout: Long,
+ probe: () => Outcome,
+ failure: (Outcome, Int, Boolean) => Unit): Unit = {
+ logInfo(description)
+ val timelimit = now() + timeout
+ var result: Outcome = Retry()
+ var current = 0L
+ var iterations = 0
+ do {
+ iterations += 1
+ result = probe()
+ if (result == Retry()) {
+ // probe says retry
+ current = now()
+ if (current> timelimit) {
+ // timeout, uprate to fail
+ result = TimedOut()
+ } else {
+ Thread.sleep(interval)
+ }
+ }
+ } while (result == Retry())
+ result match {
+ case Fail() => failure(result, iterations, false)
+ case TimedOut() => failure(result, iterations, true)
+ case _ =>
+ }
+ }
+
+ /**
+ * Convert a boolean into a success or retry outcome, that is:
+ * false is considered "retry", not a failure
+ *
+ * @param value value to probe
+ * @return
+ */
+ def outcomeFromBool(value: Boolean): Outcome = {
+ if (value) Success() else Retry()
+ }
+
+ /**
+ * From an increasing counter, compare the results and decide whether to succeed, fail or try
+ * again. Requires that if actual is greater than expected, it is a failed state.
+ *
+ * @param expected expected outcome
+ * @param actual actual value
+ */
+ def outcomeFromCounter(expected: Long, actual: Long): Outcome = {
+ if (expected == actual) {
+ Success()
+ } else if (actual < expected) {
+ Retry()
+ } else {
+ Fail()
+ }
+ }
+
+ /**
+ * From an increasing counter, compare the results and decide whether to succeed or try
+ * again. Any value equal to or greater than expected is a success. Ideal for waiting for
+ * asynchronous operations to complete
+ * @param expected expected outcome
+ * @param actual actual value
+ */
+ def outcomeAtLeast(expected: Long, actual: Long): Outcome = {
+ if (actual >= expected) Success() else Retry()
+ }
+
+ /**
+ * Curryable function to use for timeouts if something more specific is not needed
+ *
+ * @param text text mesage on timeouts
+ * @param iterations number of iterations performed
+ * @param timeout true if the event was a timeout (i.e. not a failure)
+ */
+ def timeout(text: String, iterations: Int, timeout: Boolean): Unit = {
+ fail(text)
+ }
+
+ /**
+ * a No-op on failure
+ *
+ * @param outcome outcome of the last operation
+ * @param iterations number of iterations performed
+ * @param timeout did the wait result in a timeout
+ */
+ def failure_noop(outcome: Outcome, iterations: Int, timeout: Boolean): Unit = {
+ }
+
+ /**
+ * Spin for the number of processed events to exactly match the supplied value.
+ *
+ * Fails if the timeout is exceeded
+ *
+ * @param historyService history
+ * @param expected exact number to await
+ * @param timeout timeout in millis
+ */
+ def awaitEventsProcessed(historyService: YarnHistoryService,
+ expected: Int, timeout: Long): Unit = {
+
+ def eventsProcessedCheck(): Outcome = {
+ outcomeFromCounter(expected, historyService.eventsProcessed)
+ }
+
+ def eventProcessFailure(outcome: Outcome, iterations: Int, timeout: Boolean): Unit = {
+ val eventsCount = historyService.eventsProcessed
+ val details = s"Expected $expected events" + s" actual=$eventsCount" +
+ s" after $iterations iterations; in $historyService"
+ if (timeout) {
+ val message = s"Timeout: $details"
+ logError(message)
+ fail(message)
+ } else if (expected != eventsCount) {
+ val message = s"event count mismatch; $details;"
+ logError(message)
+ fail(message)
+ fail(s"Expected $details")
+ }
+ }
+
+ spinForState("awaitEventsProcessed",
+ interval = 50,
+ timeout = timeout,
+ probe = eventsProcessedCheck,
+ failure = eventProcessFailure)
+ }
+
+ /**
+ * Spin awaiting a URL to be accessible. Useful to await a web application
+ * going live before running the tests against it
+ *
+ * @param url URL to probe
+ * @param timeout timeout in mils
+ */
+ def awaitURL(url: URL, timeout: Long): Unit = {
+ def probe(): Outcome = {
+ try {
+ url.openStream().close()
+ Success()
+ } catch {
+ case ioe: IOException => Retry()
+ }
+ }
+
+ /*
+ failure action is simply to attempt the connection without
+ catching the exception raised
+ */
+ def failure(outcome: Outcome, iterations: Int, timeout: Boolean): Unit = {
+ url.openStream().close()
+ }
+
+ spinForState(s"Awaiting a response from URL $url",
+ interval = 50, timeout = timeout, probe = probe, failure = failure)
+ }
+
+
+ /**
+ * Wait for a history service's queue to become empty
+ *
+ * @param historyService service
+ * @param timeout timeout
+ */
+ def awaitEmptyQueue(historyService: YarnHistoryService, timeout: Long): Unit = {
+
+ spinForState("awaiting empty queue",
+ interval = 50,
+ timeout = timeout,
+ probe = () => outcomeFromBool(historyService.postingQueueSize == 0),
+ failure = (_, _, _) => fail(s"queue never cleared after $timeout mS for $historyService"))
+ }
+
+ /**
+ * Await for the count of flushes in the history service to match the expected value
+ *
+ * @param historyService service
+ * @param count min number of flushes
+ * @param timeout timeout
+ */
+ def awaitFlushCount(historyService: YarnHistoryService, count: Long, timeout: Long): Unit = {
+ spinForState(s"awaiting flush count of $count",
+ interval = 50,
+ timeout = timeout,
+ probe = () => outcomeFromBool(historyService.getFlushCount >= count),
+ failure = (_, _, _) => fail(s"flush count not $count after $timeout mS in $historyService"))
+ }
+
+ /**
+ * Await the number of post events
+ * @param service service
+ * @param posts attempt count.
+ */
+ def awaitPostAttemptCount(service: YarnHistoryService, posts: Long): Unit = {
+ awaitCount(posts, TEST_STARTUP_DELAY,
+ () => service.postAttempts,
+ () => s"Post count in $service")
+ }
+
+ /**
+ * Await the number of post events
+ *
+ * @param service service
+ * @param posts attempt count.
+ */
+ def awaitPostSuccessCount(service: YarnHistoryService, posts: Long): Unit = {
+ awaitCount(posts, TEST_STARTUP_DELAY,
+ () => service.postSuccesses,
+ () => s"Post count in $service")
+ }
+
+ /**
+ * Await for the counter function to match the expected value
+ *
+ * @param expected desired count
+ * @param timeout timeout
+ * @param counter function to return an integer
+ * @param diagnostics diagnostics string evaluated on timeout
+ */
+ def awaitCount(expected: Long, timeout: Long,
+ counter: () => Long, diagnostics: () => String): Unit = {
+ spinForState(s"awaiting probe count of $expected",
+ 50, timeout,
+ () => outcomeFromCounter(expected, counter()),
+ (_, _, _) =>
+ fail(s"Expected $expected equalled ${counter()} after $timeout mS: ${diagnostics()}"))
+ }
+
+ /**
+ * Await for the counter function to match the expected value
+ *
+ * @param expected desired count
+ * @param timeout timeout
+ * @param counter function to return an integer
+ * @param diagnostics diagnostics string evaluated on timeout
+ */
+ def awaitAtLeast(expected: Long, timeout: Long,
+ counter: () => Long, diagnostics: () => String): Unit = {
+ spinForState(s"awaiting probe count of at least $expected",
+ 50, timeout,
+ () => outcomeAtLeast(expected, counter()),
+ (_, _, _) =>
+ fail(s"Expected >= $expected got ${counter()} after $timeout mS: ${diagnostics()}"))
+ }
+
+
+ /**
+ * Probe operation to wait for an empty queue
+ *
+ * @param historyService history service
+ * @param timeout timeout in milliseconds
+ * @param failOnTimeout flag -fail vs warn on timeout. Default: true
+ */
+ def awaitServiceThreadStopped(historyService: YarnHistoryService, timeout: Long,
+ failOnTimeout: Boolean = true): Unit = {
+ assertNotNull(historyService, "null historyService")
+ spinForState("awaitServiceThreadStopped",
+ interval = 50,
+ timeout = timeout,
+ probe = () => outcomeFromBool(!historyService.isPostThreadActive),
+ failure = (_, _, _) => if (failOnTimeout) {
+ fail(s"After $timeout mS, history service post thread did not finish:" +
+ s" $historyService")
+ } else {
+ logWarning(s"After $timeout mS, history service post thread did not finish:" +
+ s" $historyService")
+ })
+ }
+
+ /**
+ * Wait for a specified operation to return a list of the desired size
+ *
+ * @param expectedSize expected size of list
+ * @param message message on failure
+ * @param timeout timeout
+ * @param operation operation to create the list
+ * @tparam T list type
+ * @return the list created in the last successful operation
+ */
+ def awaitListSize[T](expectedSize: Int, message: String, timeout: Long,
+ operation: () => List[T]): List[T] = {
+ // last list fetched
+ var list: List[T] = Nil
+ def probeOperation(): Outcome = {
+ list = operation()
+ outcomeFromBool(list.size == expectedSize)
+ }
+ def failOperation(o: Outcome, i: Int, b: Boolean) = {
+ assertListSize(list, expectedSize, message)
+ }
+ spinForState(message, 50, timeout, probeOperation, failOperation)
+ list
+ }
+
+ /**
+ * Show a Spark context in a string form
+ * @param ctx context
+ * @return a string value for assertions and other messages
+ */
+ def asString(ctx: SparkContext): String = {
+ s"Spark Context ${ctx.appName} ID ${ctx.applicationId} attempts ${ctx.applicationAttemptId}"
+ }
+
+}
+
+
+
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/AbstractMockHistorySuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/AbstractMockHistorySuite.scala
new file mode 100644
index 000000000000..69a90865abc3
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/AbstractMockHistorySuite.scala
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import org.apache.hadoop.service.ServiceOperations
+import org.apache.hadoop.yarn.api.records.{ApplicationAttemptId, ApplicationId}
+import org.apache.hadoop.yarn.api.records.timeline.{TimelineEntity, TimelinePutResponse}
+import org.apache.hadoop.yarn.client.api.TimelineClient
+import org.mockito.Matchers._
+import org.mockito.Mockito._
+import org.scalatest.mock.MockitoSugar
+
+import org.apache.spark.SparkContext
+import org.apache.spark.deploy.history.yarn.YarnHistoryService
+import org.apache.spark.deploy.history.yarn.testtools.{AbstractYarnHistoryTests, TimelineServiceEnabled}
+import org.apache.spark.scheduler.cluster.SchedulerExtensionServiceBinding
+
+/**
+ * Mock histories have the timeline service enabled by default -it is
+ * a mock one though.
+ */
+class AbstractMockHistorySuite() extends AbstractYarnHistoryTests
+ with TimelineServiceEnabled with MockitoSugar {
+
+ protected var timelineClient: TimelineClient = _
+
+ protected var response: TimelinePutResponse = _
+
+ /**
+ * Set up the mock timeline client and response instances.
+ */
+ override protected def setup(): Unit = {
+ super.setup()
+ timelineClient = mock[TimelineClient]
+ response = mock[TimelinePutResponse]
+ when(timelineClient.putEntities(any(classOf[TimelineEntity]))).thenReturn(response)
+ }
+
+ override def afterEach(): Unit = {
+ ServiceOperations.stopQuietly(timelineClient)
+ super.afterEach()
+ }
+
+ /**
+ * Create and start a history service.
+ * @param sc context
+ * @param id application ID
+ * @param attemptId
+ * @return the instantiated service
+ */
+ override protected def startHistoryService(
+ sc: SparkContext,
+ id: ApplicationId,
+ attemptId: Option[ApplicationAttemptId] = None): YarnHistoryService = {
+ val service = spy(new YarnHistoryService())
+ assertNotNull(timelineClient, "timeline client")
+ doReturn(timelineClient).when(service).createTimelineClient()
+ service.start(SchedulerExtensionServiceBinding(sc, id, attemptId))
+ service
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/EventMarshallingSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/EventMarshallingSuite.scala
new file mode 100644
index 000000000000..a3dad56edd08
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/EventMarshallingSuite.scala
@@ -0,0 +1,171 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import java.io.IOException
+
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.{Logging, SparkFunSuite}
+import org.apache.spark.deploy.history.yarn.YarnHistoryService
+import org.apache.spark.deploy.history.yarn.YarnTimelineUtils._
+import org.apache.spark.deploy.history.yarn.testtools.ExtraAssertions
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+import org.apache.spark.scheduler.{AccumulableInfo, JobSucceeded, SparkListenerBlockUpdated, SparkListenerEvent, SparkListenerJobEnd, SparkListenerStageCompleted, SparkListenerStageSubmitted, SparkListenerTaskGettingResult, SparkListenerTaskStart, StageInfo, TaskInfo, TaskLocality}
+
+/**
+ * Test low-level marshalling, robustness and quality of exception messages.
+ */
+class EventMarshallingSuite extends SparkFunSuite
+ with BeforeAndAfter with Logging with ExtraAssertions {
+
+ val stageInfo = new StageInfo(12, 13, "stageinfo-1", 4, Nil, Nil, "staged info")
+
+ val taskInfo = new TaskInfo(100, 101, 102, 103, "executor", "host", TaskLocality.ANY, true)
+
+ before {
+ stageInfo.submissionTime = Some(100000)
+ stageInfo.completionTime = Some(200000)
+ stageInfo.failureReason = Some("network problems")
+ val ai = new AccumulableInfo(1, "accumulator", Some("update"), "value", false)
+ stageInfo.accumulables.put(1, ai)
+ }
+
+ test("unmarshall empty event") {
+ val event = new TimelineEvent
+ val ex = intercept[IOException] {
+ toSparkEvent(event)
+ }
+ assertExceptionMessageContains(ex, E_EMPTY_EVENTINFO)
+ }
+
+ test("unmarshall entity type") {
+ val event = new TimelineEvent
+ event.setEventType(YarnHistoryService.SPARK_EVENT_ENTITY_TYPE)
+ val ex = intercept[IOException] {
+ toSparkEvent(event)
+ }
+ assertExceptionMessageContains(ex, E_EMPTY_EVENTINFO)
+ }
+
+ test("round trip app start") {
+ val startEvent = appStartEvent(1)
+ assert(APP_USER === startEvent.sparkUser)
+ assert(APP_NAME === startEvent.appName)
+ val dest = validateRoundTrip(startEvent)
+ assert(startEvent.time === dest.time )
+ assert(startEvent.sparkUser === dest.sparkUser )
+ assert(APP_NAME === dest.appName)
+ }
+
+ test("round trip app end") {
+ validateRoundTrip(appStopEvent(1))
+ }
+
+ test("SparkListenerStageSubmitted") {
+ val src = new SparkListenerStageSubmitted(stageInfo)
+ val dest = roundTrip(src)
+ assert(isEqual(stageInfo, dest.stageInfo))
+ }
+
+ test("SparkListenerStageCompleted") {
+ val src = new SparkListenerStageCompleted(stageInfo)
+ val dest = roundTrip(src)
+ assert(isEqual(stageInfo, dest.stageInfo))
+ }
+
+ test("SparkListenerTaskStart") {
+ val src = new SparkListenerTaskStart(1, 2, taskInfo)
+ val dest = roundTrip(src)
+ assert(isEqual(taskInfo, dest.taskInfo))
+ }
+
+ test("SparkListenerTaskGettingResult") {
+ val src = new SparkListenerTaskGettingResult(taskInfo)
+ val dest = roundTrip(src)
+ assert(isEqual(taskInfo, dest.taskInfo))
+ }
+
+ test("SparkListenerJobEnd") {
+ val endTime = 3000L
+ val id = 3
+ val result = JobSucceeded
+ val src = new SparkListenerJobEnd(id, endTime, result)
+ val dest = roundTrip(src)
+ assert(endTime === dest.time)
+ assert(id === dest.jobId)
+ assert(result === dest.jobResult)
+ }
+
+ test("SparkListenerBlockUpdated is ignored") {
+ assert(toTimelineEvent(new SparkListenerBlockUpdated(null), 0).isEmpty)
+ }
+
+ def validateRoundTrip[T <: SparkListenerEvent](sparkEvt: T): T = {
+ val trip = roundTrip(sparkEvt)
+ assertResult(sparkEvt) {
+ trip
+ }
+ trip
+ }
+
+ /**
+ * Marshall then unmarshall a spark event.
+ *
+ * @param src source
+ * @return a new spark event built from the marshalled JSON value.
+ */
+ private def roundTrip[T <: SparkListenerEvent ](src: T): T = {
+ val event = toSparkEvent(toTimelineEvent(src, 100).get)
+ event.asInstanceOf[T]
+ }
+
+ /**
+ * Task info equality; does not check accumulables.
+ *
+ * @param l left item
+ * @param r right item
+ * @return true if the values are equal
+ */
+ def isEqual(l: TaskInfo, r: TaskInfo) : Boolean = {
+ l.taskId == r.taskId &&
+ l.index == r.index &&
+ l.attemptNumber == r.attemptNumber &&
+ l.executorId == r.executorId &&
+ l.host == r.host &&
+ l.speculative == r.speculative &&
+ l.taskLocality == r.taskLocality &&
+ l.gettingResultTime == r.gettingResultTime &&
+ l.finishTime == r.finishTime &&
+ l.failed == r.failed &&
+ l.accumulables.size == r.accumulables.size
+ }
+
+ def isEqual(l: StageInfo, r: StageInfo): Boolean = {
+ l.stageId == r.stageId &&
+ l.name == r.name &&
+ l.attemptId == r.attemptId &&
+ l.numTasks == r.numTasks &&
+ l.details == r.details &&
+ l.submissionTime == r.submissionTime &&
+ l.completionTime == r.completionTime &&
+ l.failureReason == r.failureReason &&
+ l.accumulables.size == r.accumulables.size
+ }
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/HistoryServiceInstantiationSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/HistoryServiceInstantiationSuite.scala
new file mode 100644
index 000000000000..7ef1a2970541
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/HistoryServiceInstantiationSuite.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.history.yarn.YarnHistoryService
+import org.apache.spark.deploy.history.yarn.testtools.AbstractYarnHistoryTests
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+import org.apache.spark.scheduler.cluster.{SchedulerExtensionServiceBinding, SchedulerExtensionServices}
+
+/**
+ * Test the integration with [[SchedulerExtensionServices]].
+ */
+class HistoryServiceInstantiationSuite extends AbstractYarnHistoryTests {
+
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ super.setupConfiguration(sparkConf)
+ sparkConf.set(SchedulerExtensionServices.SPARK_YARN_SERVICES, YarnHistoryService.CLASSNAME)
+ }
+
+ test("Contains History Service") {
+ val services = new SchedulerExtensionServices
+ try {
+ services.start(SchedulerExtensionServiceBinding(sc, applicationId))
+ val serviceList = services.getServices
+ assert(serviceList.nonEmpty, "empty service list")
+ val Seq(history) = serviceList
+ val historyService = history.asInstanceOf[YarnHistoryService]
+ assert(historyService.serviceState === YarnHistoryService.StartedState)
+ services.stop()
+ assert(historyService.serviceState === YarnHistoryService.StoppedState)
+ } finally {
+ services.stop()
+ }
+ }
+
+}
+
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/HistoryWithDisabledTimelineSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/HistoryWithDisabledTimelineSuite.scala
new file mode 100644
index 000000000000..b14be6efb6a4
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/HistoryWithDisabledTimelineSuite.scala
@@ -0,0 +1,75 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import org.apache.spark.deploy.history.yarn.YarnHistoryService
+import org.apache.spark.deploy.history.yarn.YarnHistoryService._
+import org.apache.spark.deploy.history.yarn.testtools.AbstractYarnHistoryTests
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+import org.apache.spark.scheduler.cluster.SchedulerExtensionServiceBinding
+
+/**
+ * Test that with the timeline service disabled, public operations degrade gracefully.
+ */
+class HistoryWithDisabledTimelineSuite extends AbstractYarnHistoryTests {
+
+ test("BasicLifecycle") {
+ val service = new YarnHistoryService()
+ // verify that the string operator does not fail
+ service.toString()
+
+ service.start(SchedulerExtensionServiceBinding(sc, applicationId, Some(attemptId)))
+ assert(StartedState === service.serviceState, "not stopped : $service")
+ assert(!service.bondedToATS, s"service is bonded to ats $service")
+ assert(!service.listening, s"service is listening $service")
+ assertResult(null, s"service address : $service") {
+ service.timelineWebappAddress
+ }
+ intercept[Exception] {
+ service.timelineClient
+ }
+ assert(!service.isPostThreadActive, s"service post thread active: $service")
+
+ // verify that the string operator does not fail
+ service.toString()
+ service.stop()
+ assert(StoppedState === service.serviceState, "not stopped : $service")
+ // verify that the string operator does not fail
+ service.toString()
+ }
+
+ test("QueueAndFlush") {
+ val service = new YarnHistoryService()
+ try {
+ service.start(SchedulerExtensionServiceBinding(sc, applicationId, Some(attemptId)))
+ service.enqueue(appStartEvent())
+ service.enqueue(appStopEvent())
+
+ assert(0 === service.eventsQueued, "queue")
+
+ service.asyncFlush()
+ assert(0 === service.getFlushCount, "flush count")
+
+ service.stop()
+ assert(0 === service.getFlushCount, "flush count")
+ } finally {
+ service.stop()
+ }
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockBatchingTimelinePostSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockBatchingTimelinePostSuite.scala
new file mode 100644
index 000000000000..94dd20faa6dd
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockBatchingTimelinePostSuite.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity
+import org.mockito.Matchers._
+import org.mockito.Mockito._
+
+import org.apache.spark.deploy.history.yarn.testtools.TimelineSingleEntryBatchSize
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+
+/**
+ * Mock tests with Batch size 1.
+ */
+class MockBatchingTimelinePostSuite extends AbstractMockHistorySuite
+ with TimelineSingleEntryBatchSize {
+
+ test("retry upload on failure") {
+ describe("mock failures, verify retry count incremented")
+ // timeline client to throw an RTE on the first put
+ when(timelineClient.putEntities(any(classOf[TimelineEntity])))
+ .thenThrow(new RuntimeException("triggered"))
+ .thenReturn(response)
+
+ val (service, eventsPosted) = postEvents(sc)
+ // now await some retries asynchronously
+ awaitAtLeast(2, TEST_STARTUP_DELAY,
+ () => service.postAttempts,
+ () => s"Post count in $service")
+ service.stop()
+ awaitServiceThreadStopped(service, TEST_STARTUP_DELAY)
+ // there should have been three flushed
+ assert(eventsPosted === service.getFlushCount, s"expected $eventsPosted flushed for $service" )
+ verify(timelineClient, times(service.postAttempts.toInt))
+ .putEntities(any(classOf[TimelineEntity]))
+ }
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryBulkPostingSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryBulkPostingSuite.scala
new file mode 100644
index 000000000000..28f33fba2373
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryBulkPostingSuite.scala
@@ -0,0 +1,84 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import java.io.IOException
+
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity
+import org.mockito.Matchers._
+import org.mockito.Mockito._
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.history.yarn.YarnEventListener
+import org.apache.spark.deploy.history.yarn.YarnHistoryService._
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+import org.apache.spark.scheduler.SparkListenerJobStart
+
+/**
+ * Mock event posting
+ */
+class MockHistoryBulkPostingSuite extends AbstractMockHistorySuite {
+
+ val batchSize = 5
+ val queueLimit = 5
+
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ val conf = super.setupConfiguration(sparkConf)
+ conf.set(BATCH_SIZE, batchSize.toString())
+ conf.set(POST_EVENT_LIMIT, queueLimit.toString)
+ conf.set(POST_RETRY_INTERVAL, "0ms")
+ }
+
+ test("Massive Event Posting") {
+ describe("Post many events to a failing")
+ // timeline client to throw an exception on every POST
+
+ when(timelineClient.putEntities(any(classOf[TimelineEntity])))
+ .thenThrow(new IOException("triggered"))
+
+ val service = startHistoryService(sc)
+ try {
+ val listener = new YarnEventListener(sc, service)
+ // start
+ listener.onApplicationStart(applicationStart)
+ // post many more events
+ 1 to (batchSize * 2 * queueLimit ) foreach { t =>
+ listener.onJobStart(new SparkListenerJobStart(1, t, Nil))
+ }
+ // events dropped
+ awaitAtLeast(batchSize, TEST_STARTUP_DELAY,
+ () => service.eventsDropped,
+ () => service.toString())
+
+ // posts failed
+ awaitAtLeast(10, SERVICE_SHUTDOWN_DELAY,
+ () => service.postFailures,
+ () => service.toString())
+
+ // now trigger a service shutdown with the blocking queue
+ describe("Service stop")
+ service.stop()
+ awaitServiceThreadStopped(service, SERVICE_SHUTDOWN_DELAY)
+ logDebug(s"$service")
+ } finally {
+ logDebug("stopping service in finally() clause")
+ service.stop()
+ }
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryEventPostingSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryEventPostingSuite.scala
new file mode 100644
index 000000000000..18f35284e449
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryEventPostingSuite.scala
@@ -0,0 +1,170 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import java.util.concurrent.atomic.AtomicBoolean
+
+import org.apache.hadoop.yarn.api.records.timeline.{TimelineEntity, TimelinePutResponse}
+import org.mockito.Matchers._
+import org.mockito.Mockito._
+import org.mockito.invocation.InvocationOnMock
+import org.mockito.stubbing.Answer
+
+import org.apache.spark.deploy.history.yarn.YarnEventListener
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+
+/**
+ * Mock event posting.
+ */
+class MockHistoryEventPostingSuite extends AbstractMockHistorySuite {
+
+ /*
+ * Make sure the low-level stuff the other tests depend on is there
+ */
+ test("Timeline client") {
+ describe("low-level timeline client test")
+ assert(response === timelineClient.putEntities(new TimelineEntity))
+ verify(timelineClient).putEntities(any(classOf[TimelineEntity]))
+ }
+
+ test("Event Queueing") {
+ describe("event queueing")
+ val (history, eventsPosted) = postEvents(sc)
+ awaitEventsProcessed(history, eventsPosted, TEST_STARTUP_DELAY)
+ }
+
+ test("batch processing of Spark listener events") {
+ val (historyService, _) = postEvents(sc)
+ verify(timelineClient, times(historyService.postAttempts.toInt))
+ .putEntities(any(classOf[TimelineEntity]))
+ }
+
+ test("PostEventsNoServiceStop") {
+ describe("verify that events are pushed on any triggered flush," +
+ " even before a service is stopped")
+ val service = startHistoryService(sc)
+ try {
+ val listener = new YarnEventListener(sc, service)
+ listener.onApplicationStart(applicationStart)
+ service.asyncFlush()
+ awaitEventsProcessed(service, 1, TEST_STARTUP_DELAY)
+ awaitFlushCount(service, 1, TEST_STARTUP_DELAY)
+ awaitPostAttemptCount(service, 1)
+ logDebug(s"$service")
+ verify(timelineClient, times(1)).putEntities(any(classOf[TimelineEntity]))
+ } finally {
+ logDebug("stopping service in finally() clause")
+ service.stop()
+ }
+ }
+
+ /**
+ * This is a convoluted little test designed to verify something: stopping
+ * the service while it is waiting for something to happen in the thread posting
+ * service will cause the thread to complete successfully.
+ *
+ * That is: even though the YARN code may potentially swallow interrupted exceptions,
+ * once it returns, the exit flag is picked up and the post thread switches
+ * into fast shutdown mode, attempting to post any remaining threads.
+ */
+ test("PostEventsBlockingOperation") {
+ describe("verify that events are pushed on any triggered flush," +
+ " even before a service is stopped")
+ val entered = new AtomicBoolean(false)
+ val exit = new AtomicBoolean(false)
+
+ /**
+ * Set a flag ang then notify the listeners
+ * @param b atomic bool flag to set
+ */
+ def setAndNotify(b: AtomicBoolean): Unit = {
+ b.synchronized {
+ b.set(true)
+ b.notify()
+ }
+ }
+
+ // wait for a boolean to be set; interrupts are discarded.
+ def waitForSet(b: AtomicBoolean, timeout: Long): Unit = {
+ b.synchronized {
+ while (!b.get()) {
+ try {
+ b.wait(timeout)
+ if (!b.get()) {
+ fail("post operation never started")
+ }
+ } catch {
+ case irq: InterruptedException =>
+ case ex: RuntimeException => throw ex
+ }
+ }
+ }
+ }
+
+ // Mockito answer which doesn't return until the `exit` flag is set,
+ // and which sets the `entered` flag on entry.
+ // this is designed to synchronize the posting thread with the test runner thread
+ class delayedAnswer extends Answer[TimelinePutResponse]() {
+
+
+ override def answer(invocation: InvocationOnMock): TimelinePutResponse = {
+ // flag the operation has started
+ if (!exit.get()) {
+ logDebug("setting `entered` flag")
+ setAndNotify(entered)
+ logDebug("waiting for `exit` flag")
+ exit.synchronized {
+ exit.wait(TEST_STARTUP_DELAY)
+ }
+ }
+ new TimelinePutResponse
+ }
+ }
+ when(timelineClient.putEntities(any(classOf[TimelineEntity])))
+ .thenAnswer(new delayedAnswer())
+
+ val service = startHistoryService(sc)
+ try {
+ val listener = new YarnEventListener(sc, service)
+ listener.onApplicationStart(applicationStart)
+ awaitPostAttemptCount(service, 1)
+ setAndNotify(entered)
+ entered.synchronized {
+ if (!entered.get()) {
+ logDebug("waiting for `entered` flag")
+ entered.wait(TEST_STARTUP_DELAY)
+ if (!entered.get()) {
+ fail("post operation never started")
+ }
+ }
+ }
+ // trigger the stop process. The interrupt will be probably be lost
+ logDebug("stopping service")
+ service.stop()
+ logDebug("setting `exit` flag")
+ setAndNotify(exit)
+ awaitServiceThreadStopped(service, SERVICE_SHUTDOWN_DELAY)
+ logDebug(s"$service")
+ } finally {
+ setAndNotify(exit)
+ logDebug("stopping service in finally() clause")
+ service.stop()
+ }
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryFlushingSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryFlushingSuite.scala
new file mode 100644
index 000000000000..a893e8d2acc0
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryFlushingSuite.scala
@@ -0,0 +1,80 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity
+import org.mockito.Matchers._
+import org.mockito.Mockito._
+import org.scalatest.{BeforeAndAfter, Matchers}
+
+import org.apache.spark.Logging
+import org.apache.spark.deploy.history.yarn.YarnEventListener
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+import org.apache.spark.scheduler.SparkListenerUnpersistRDD
+
+/**
+ * Tests to verify that timeline operations happen even before services are closed.
+ *
+ * There's an async queue involved here, so the tests spin until a state is met or not.
+ */
+class MockHistoryFlushingSuite extends AbstractMockHistorySuite
+ with BeforeAndAfter with Matchers with Logging {
+
+ test("PostEventsNoServiceStop") {
+ describe("verify that events are pushed on any triggered flush," +
+ " even before a service is stopped")
+ val service = startHistoryService(sc)
+ try {
+ assert(service.timelineServiceEnabled, s"no timeline service in $service")
+ service.timelineClient
+ service.createTimelineClient()
+ val listener = new YarnEventListener(sc, service)
+ listener.onApplicationStart(applicationStart)
+ service.asyncFlush()
+ awaitPostAttemptCount(service, 1)
+ verify(timelineClient, times(1)).putEntities(any(classOf[TimelineEntity]))
+ } finally {
+ service.stop()
+ }
+ }
+
+ test("PostEventsWithServiceStop") {
+ describe("verify that events are pushed on service stop")
+ val service = startHistoryService(sc)
+ try {
+ service.timelineClient
+ service.createTimelineClient()
+ val listener = new YarnEventListener(sc, service)
+ listener.onApplicationStart(applicationStart)
+ awaitPostAttemptCount(service, 1)
+ verify(timelineClient, times(1)).putEntities(any(classOf[TimelineEntity]))
+ listener.onUnpersistRDD(SparkListenerUnpersistRDD(1))
+ // expecting two events
+ awaitPostAttemptCount(service, 1)
+
+ // now stop the service and await the final post
+ service.stop()
+ awaitServiceThreadStopped(service, TEST_STARTUP_DELAY)
+ verify(timelineClient, times(2)).putEntities(any(classOf[TimelineEntity]))
+ } finally {
+ logDebug(s"teardown of $service")
+ service.stop()
+ }
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryServiceLifecycleSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryServiceLifecycleSuite.scala
new file mode 100644
index 000000000000..885791b1ef7f
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/MockHistoryServiceLifecycleSuite.scala
@@ -0,0 +1,92 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.history.yarn.YarnHistoryService
+import org.apache.spark.deploy.history.yarn.YarnHistoryService._
+import org.apache.spark.deploy.history.yarn.testtools.{ContextSetup, HistoryServiceListeningToSparkContext}
+import org.apache.spark.deploy.history.yarn.testtools.YarnTestUtils._
+import org.apache.spark.scheduler.cluster.SchedulerExtensionServiceBinding
+
+class MockHistoryServiceLifecycleSuite
+ extends AbstractMockHistorySuite
+ with ContextSetup
+ with HistoryServiceListeningToSparkContext {
+
+ /**
+ * Set the batch size to 2, purely so that we can trace its path through
+ * the configuration system.
+ */
+ override def setupConfiguration(sparkConf: SparkConf): SparkConf = {
+ super.setupConfiguration(sparkConf).set(BATCH_SIZE, "2")
+ }
+
+ /*
+ * Test service lifecycle ops and that stop() is re-entrant
+ */
+ test("Service Lifecycle") {
+ describe("service lifecycle operations")
+
+ assertResult("2", s"batch size in context") {
+ sc.conf.get(BATCH_SIZE)
+ }
+
+ assertResult("true", s"listening flag") {
+ sc.conf.get(REGISTER_LISTENER)
+ }
+
+ val service = startHistoryService(sc)
+ assertResult(StartedState, "not started") {
+ service.serviceState
+ }
+ assertResult(2, s"batch size in $service") {
+ service.batchSize
+ }
+ assertResult(true, s"listen flag in $service") {
+ service.listening
+ }
+
+ service.stop()
+ assertResult(StoppedState, s"not stopped: $service") {
+ service.serviceState
+ }
+
+ // and expect an attempt to start again to fail
+ intercept[IllegalArgumentException] {
+ service.start(SchedulerExtensionServiceBinding(sc, applicationId, None))
+ }
+ // repeated stop() is harmless
+ service.stop()
+ }
+
+ test("ServiceStartArguments1") {
+ val service = new YarnHistoryService()
+ intercept[IllegalArgumentException] {
+ service.start(SchedulerExtensionServiceBinding(null, applicationId, None))
+ }
+ }
+
+ test("ServiceStartArguments2") {
+ val service = new YarnHistoryService()
+ intercept[IllegalArgumentException] {
+ service.start(SchedulerExtensionServiceBinding(sc, null, None))
+ }
+ }
+
+}
diff --git a/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/YarnTimelineUtilsSuite.scala b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/YarnTimelineUtilsSuite.scala
new file mode 100644
index 000000000000..0f4ab0ff04d2
--- /dev/null
+++ b/yarn/src/history/test/scala/org/apache/spark/deploy/history/yarn/unit/YarnTimelineUtilsSuite.scala
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.deploy.history.yarn.unit
+
+import java.net.{NoRouteToHostException, URI}
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.{Logging, SparkFunSuite}
+import org.apache.spark.deploy.history.yarn.YarnTimelineUtils._
+import org.apache.spark.deploy.history.yarn.testtools.ExtraAssertions
+
+/**
+ * Tests of methods in [[org.apache.spark.deploy.history.yarn.YarnTimelineUtils]].
+ */
+class YarnTimelineUtilsSuite extends SparkFunSuite
+ with BeforeAndAfter with Logging with ExtraAssertions {
+
+ test("verifyNoHost") {
+ intercept[NoRouteToHostException] {
+ validateEndpoint(new URI("http://0.0.0.0:8080/ws"))
+ }
+ }
+
+ test("verifyNoPort") {
+ intercept[NoRouteToHostException] {
+ validateEndpoint(new URI("http://127.0.1.1:0/ws"))
+ }
+ }
+
+ test("verifyValid") {
+ validateEndpoint(new URI("http://127.0.1.1:8080/ws"))
+ }
+
+}