Skip to content

[SPARK-13371] [CORE] TaskSetManager.dequeueSpeculativeTask compares Option[String] and String directly. #11253

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

[SPARK-13371] [CORE] TaskSetManager.dequeueSpeculativeTask compares Option[String] and String directly. #11253

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1135f21
Fix some comparisons between unequal types that cause IJ warnings and…
srowen Feb 18, 2016
bd4b3fb
parentRDD -> parentDependency
srowen Feb 18, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/deploy/FaultToleranceTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,7 @@ private object FaultToleranceTest extends App with Logging {
val f = Future {
try {
val res = sc.parallelize(0 until 10).collect()
assertTrue(res.toList == (0 until 10))
assertTrue(res.toList == (0 until 10).toList)
true
} catch {
case e: Exception =>
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -338,7 +338,7 @@ private[spark] class TaskSetManager(
if (TaskLocality.isAllowed(locality, TaskLocality.RACK_LOCAL)) {
for (rack <- sched.getRackForHost(host)) {
for (index <- speculatableTasks if canRunOnHost(index)) {
Copy link
Contributor

@markhamstra markhamstra Feb 18, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit, but I'd actually refactor this as a single for:

for {
  rack <- sched.getRackForHost(host)
  index <- speculatableTasks if canRunOnHost(index)
} {
  val racks = tasks(index).preferredLocations.map(_.host).flatMap(sched.getRackForHost)
  if (racks.contains(rack)) {
    speculatableTasks -= index
    return Some((index, TaskLocality.RACK_LOCAL))
  }
}

Copy link
Contributor

@andrewor14 andrewor14 Feb 18, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not a big deal but I think that can happen outside of this patch since it's just cleanup

Copy link
Contributor

@markhamstra markhamstra Feb 18, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yup, it's style, not substance

val racks = tasks(index).preferredLocations.map(_.host).map(sched.getRackForHost)
val racks = tasks(index).preferredLocations.map(_.host).flatMap(sched.getRackForHost)
if (racks.contains(rack)) {
speculatableTasks -= index
return Some((index, TaskLocality.RACK_LOCAL))
Expand Down
4 changes: 2 additions & 2 deletions core/src/test/scala/org/apache/spark/CheckpointSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ trait RDDCheckpointTester { self: SparkFunSuite =>
// Generate the final RDD using given RDD operation
val baseRDD = generateFatRDD()
val operatedRDD = op(baseRDD)
val parentRDD = operatedRDD.dependencies.headOption.orNull
val parentDependency = operatedRDD.dependencies.headOption.orNull
val rddType = operatedRDD.getClass.getSimpleName
val numPartitions = operatedRDD.partitions.length

Expand Down Expand Up @@ -82,7 +82,7 @@ trait RDDCheckpointTester { self: SparkFunSuite =>
}

// Test whether dependencies have been changed from its earlier parent RDD
assert(operatedRDD.dependencies.head.rdd != parentRDD)
assert(operatedRDD.dependencies.head != parentDependency)

// Test whether the partitions have been changed from its earlier partitions
assert(operatedRDD.partitions.toList != partitionsBeforeCheckpoint.toList)
Expand Down
4 changes: 2 additions & 2 deletions core/src/test/scala/org/apache/spark/PartitioningSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,8 +163,8 @@ class PartitioningSuite extends SparkFunSuite with SharedSparkContext with Priva
val hashP2 = new HashPartitioner(2)
assert(rangeP2 === rangeP2)
assert(hashP2 === hashP2)
assert(hashP2 != rangeP2)
assert(rangeP2 != hashP2)
assert(hashP2 !== rangeP2)
assert(rangeP2 !== hashP2)
}

test("partitioner preservation") {
Expand Down
6 changes: 5 additions & 1 deletion core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -289,7 +289,11 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser with Matchers with B
JInt(stageId) <- stage \ "stageId"
JInt(attemptId) <- stage \ "attemptId"
} {
val exp = if (attemptId == 0 && stageId == 1) StageStatus.FAILED else StageStatus.COMPLETE
val exp = if (attemptId.toInt == 0 && stageId.toInt == 1) {
StageStatus.FAILED
} else {
StageStatus.COMPLETE
}
status should be (exp.name())
}

Expand Down