Skip to content

Commit

Permalink
style
Browse files Browse the repository at this point in the history
  • Loading branch information
squito committed Apr 27, 2015
1 parent a9c5cf1 commit a325563
Show file tree
Hide file tree
Showing 9 changed files with 43 additions and 59 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -768,7 +768,6 @@ private[master] class Master(
if (inProgressExists) {
// Event logging is enabled for this application, but the application is still in progress
logWarning(s"Application $appName is still in progress, it may be terminated abnormally.")
return None
}

val (eventLogFile, status) = if (inProgressExists) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,7 @@
*/
package org.apache.spark.status.api.v1

import java.util
import java.util.Date
import java.util.{Arrays, Date, List => JList}
import javax.ws.rs._
import javax.ws.rs.core.MediaType

Expand All @@ -32,16 +31,15 @@ private[v1] class AllJobsResource(uiRoot: UIRoot) {
@GET
def jobsList(
@PathParam("appId") appId: String,
@QueryParam("status") statuses: java.util.List[JobExecutionStatus]
@QueryParam("status") statuses: JList[JobExecutionStatus]
): Seq[JobData] = {
uiRoot.withSparkUI(appId) { ui =>
val statusToJobs: Seq[(JobExecutionStatus, Seq[JobUIData])] =
AllJobsResource.getStatusToJobs(ui)
val adjStatuses: util.List[JobExecutionStatus] = {
val adjStatuses: JList[JobExecutionStatus] = {
if (statuses.isEmpty) {
java.util.Arrays.asList(JobExecutionStatus.values(): _*)
}
else {
Arrays.asList(JobExecutionStatus.values(): _*)
} else {
statuses
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/
package org.apache.spark.status.api.v1

import java.util.Date
import java.util.{Arrays, Date, List => JList}
import javax.ws.rs.{GET, PathParam, Produces, QueryParam}
import javax.ws.rs.core.MediaType

Expand All @@ -33,14 +33,14 @@ private[v1] class AllStagesResource(uiRoot: UIRoot) {
@GET
def stageList(
@PathParam("appId") appId: String,
@QueryParam("status") statuses: java.util.List[StageStatus]
@QueryParam("status") statuses: JList[StageStatus]
): Seq[StageData] = {
uiRoot.withSparkUI(appId) { ui =>
val listener = ui.jobProgressListener
val stageAndStatus = AllStagesResource.stagesAndStatus(ui)
val adjStatuses = {
if (statuses.isEmpty()) {
java.util.Arrays.asList(StageStatus.values(): _*)
Arrays.asList(StageStatus.values(): _*)
} else {
statuses
}
Expand Down Expand Up @@ -279,5 +279,4 @@ private[v1] object AllStagesResource {
recordsWritten = internal.shuffleRecordsWritten
)
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,7 @@
*/
package org.apache.spark.status.api.v1

import java.util.Date
import java.util.{Arrays, List => JList}
import java.util.{Arrays, Date, List => JList}
import javax.ws.rs.{DefaultValue, GET, Produces, QueryParam}
import javax.ws.rs.core.MediaType

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import scala.util.Try
private[v1] class SimpleDateParam(val originalValue: String) {
val timestamp: Long = {
SimpleDateParam.formats.collectFirst {
case fmt if Try{ fmt.parse(originalValue) }.isSuccess =>
case fmt if Try(fmt.parse(originalValue)).isSuccess =>
fmt.parse(originalValue).getTime()
}.getOrElse(
throw new WebApplicationException(
Expand Down
12 changes: 1 addition & 11 deletions core/src/main/scala/org/apache/spark/status/api/v1/api.scala
Original file line number Diff line number Diff line change
Expand Up @@ -220,7 +220,6 @@ class OutputMetricDistributions(
val recordsWritten: IndexedSeq[Double]
)


class ShuffleReadMetricDistributions(
val readBytes: IndexedSeq[Double],
val readRecords: IndexedSeq[Double],
Expand All @@ -241,13 +240,4 @@ class AccumulableInfo (
val id: Long,
val name: String,
val update: Option[String],
val value: String) {

override def equals(other: Any): Boolean = other match {
case acc: AccumulableInfo =>
this.id == acc.id && this.name == acc.name &&
this.value == acc.value
case _ => false
}
}

val value: String)
3 changes: 1 addition & 2 deletions core/src/main/scala/org/apache/spark/ui/SparkUI.scala
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,7 @@ package org.apache.spark.ui

import java.util.Date

import org.apache.spark.status.api.v1.{ApplicationInfo, JsonRootResource}
import org.apache.spark.status.api.v1.UIRoot
import org.apache.spark.status.api.v1.{ApplicationInfo, JsonRootResource, UIRoot}
import org.apache.spark.{Logging, SecurityManager, SparkConf, SparkContext}
import org.apache.spark.scheduler._
import org.apache.spark.storage.StorageStatusListener
Expand Down
30 changes: 15 additions & 15 deletions core/src/main/scala/org/apache/spark/ui/exec/ExecutorsPage.scala
Original file line number Diff line number Diff line change
Expand Up @@ -28,21 +28,21 @@ import org.apache.spark.util.Utils

// This isn't even used anymore -- but we need to keep it b/c of a MiMa false positive
private[ui] case class ExecutorSummaryInfo(
id: String,
hostPort: String,
rddBlocks: Int,
memoryUsed: Long,
diskUsed: Long,
activeTasks: Int,
failedTasks: Int,
completedTasks: Int,
totalTasks: Int,
totalDuration: Long,
totalInputBytes: Long,
totalShuffleRead: Long,
totalShuffleWrite: Long,
maxMemory: Long,
executorLogs: Map[String, String])
id: String,
hostPort: String,
rddBlocks: Int,
memoryUsed: Long,
diskUsed: Long,
activeTasks: Int,
failedTasks: Int,
completedTasks: Int,
totalTasks: Int,
totalDuration: Long,
totalInputBytes: Long,
totalShuffleRead: Long,
totalShuffleWrite: Long,
maxMemory: Long,
executorLogs: Map[String, String])


private[ui] class ExecutorsPage(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ class HistoryServerSuite extends FunSuite with BeforeAndAfter with Matchers with
server.bind()
port = server.boundPort
}

def stop(): Unit = {
server.stop()
}
Expand Down Expand Up @@ -102,27 +103,26 @@ class HistoryServerSuite extends FunSuite with BeforeAndAfter with Matchers with
"stage with accumulable json" -> "applications/local-1426533911241/stages/0/0",
"rdd list storage json" -> "applications/local-1422981780767/storage/rdd",
"one rdd storage json" -> "applications/local-1422981780767/storage/rdd/0"
// TODO multi-attempt stages
)

// run a bunch of characterization tests -- just verify the behavior is the same as what is saved
// in the test resource folder
cases.foreach { case (name, path) =>
test(name) {
val (code, jsonOpt, errOpt) = getContentAndCode(path)
code should be (HttpServletResponse.SC_OK)
jsonOpt should be ('defined)
errOpt should be (None)
val json = jsonOpt.get
val exp = IOUtils.toString(new FileInputStream(
new File(expRoot, path + "/json_expectation")))
// compare the ASTs so formatting differences don't cause failures
import org.json4s._
import org.json4s.jackson.JsonMethods._
val jsonAst = parse(json)
val expAst = parse(exp)
assertValidDataInJson(jsonAst, expAst)
}
test(name) {
val (code, jsonOpt, errOpt) = getContentAndCode(path)
code should be (HttpServletResponse.SC_OK)
jsonOpt should be ('defined)
errOpt should be (None)
val json = jsonOpt.get
val exp = IOUtils.toString(new FileInputStream(
new File(expRoot, path + "/json_expectation")))
// compare the ASTs so formatting differences don't cause failures
import org.json4s._
import org.json4s.jackson.JsonMethods._
val jsonAst = parse(json)
val expAst = parse(exp)
assertValidDataInJson(jsonAst, expAst)
}
}

test("security") {
Expand Down

0 comments on commit a325563

Please sign in to comment.