Skip to content

Change the maxTaskFailures depending on property #186

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

Merged
suranjan merged 2 commits into from
Dec 6, 2019
Merged

Change the maxTaskFailures depending on property #186

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a746446
Change the maxTaskFailures depending on property
suranjan Nov 28, 2019
f5220b4
Refactored to add constant
suranjan Dec 3, 2019
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
15 changes: 14 additions & 1 deletion core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,8 @@ private[spark] class TaskSchedulerImpl(
// How often to check for speculative tasks
val SPECULATION_INTERVAL_MS = conf.getTimeAsMs("spark.speculation.interval", "100ms")

val SNAPPY_WRITE_RETRY_PROP = "snappydata.maxRetryAttemptsForWrite"

// Duplicate copies of a task will only be launched if the original copy has been running for
// at least this amount of time. This is to avoid the overhead of launching speculative copies
// of tasks that are very short.
Expand Down Expand Up @@ -194,7 +196,18 @@ private[spark] class TaskSchedulerImpl(
val tasks = taskSet.tasks
logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
this.synchronized {
val manager = createTaskSetManager(taskSet, maxTaskFailures)
val maxRetryAttemptsForWrite = taskSet.properties.
getProperty(SNAPPY_WRITE_RETRY_PROP)

logInfo("The maxRetryAttemptsForWrite is set to " + maxRetryAttemptsForWrite +
"maxTaskFailure " + maxTaskFailures)
val maxRetryAttempts = if (maxRetryAttemptsForWrite != null) {
maxRetryAttemptsForWrite.toInt
} else {
maxTaskFailures
}

val manager = createTaskSetManager(taskSet, maxRetryAttempts)
val stage = taskSet.stageId
val stageTaskSets =
taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
Expand Down