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.
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
RestartWithBackOff delay cancel to wait for failure #24795
Merged
Merged
Changes from all commits
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -5,11 +5,15 @@ | |
package akka.stream.scaladsl | ||
|
||
import akka.NotUsed | ||
import akka.annotation.ApiMayChange | ||
import akka.pattern.BackoffSupervisor | ||
import akka.stream.Attributes.Attribute | ||
import akka.stream._ | ||
import akka.stream.stage.{ GraphStage, InHandler, OutHandler, TimerGraphStageLogicWithLogging } | ||
import akka.stream.impl.fusing.GraphStages.SimpleLinearGraphStage | ||
import akka.stream.scaladsl.RestartWithBackoffFlow.Delay | ||
import akka.stream.stage._ | ||
|
||
import scala.concurrent.duration.FiniteDuration | ||
import scala.concurrent.duration._ | ||
|
||
/** | ||
* A RestartSource wraps a [[Source]] that gets restarted when it completes or fails. | ||
|
@@ -363,17 +367,25 @@ private final class RestartWithBackoffFlow[In, Out]( | |
val out = Outlet[Out]("RestartWithBackoffFlow.out") | ||
|
||
override def shape = FlowShape(in, out) | ||
|
||
override def createLogic(inheritedAttributes: Attributes) = new RestartWithBackoffLogic( | ||
"Flow", shape, minBackoff, maxBackoff, randomFactor, onlyOnFailures, maxRestarts) { | ||
val delay = inheritedAttributes.get[Delay](Delay(50.millis)).duration | ||
|
||
var activeOutIn: Option[(SubSourceOutlet[In], SubSinkInlet[Out])] = None | ||
|
||
override protected def logSource = self.getClass | ||
|
||
override protected def startGraph() = { | ||
val sourceOut = createSubOutlet(in) | ||
val sinkIn = createSubInlet(out) | ||
Source.fromGraph(sourceOut.source).via(flowFactory()).runWith(sinkIn.sink)(subFusingMaterializer) | ||
val sourceOut: SubSourceOutlet[In] = createSubOutlet(in) | ||
val sinkIn: SubSinkInlet[Out] = createSubInlet(out) | ||
|
||
Source.fromGraph(sourceOut.source) | ||
// Temp fix while waiting cause of cancellation. See #23909 | ||
.via(RestartWithBackoffFlow.delayCancellation[In](delay)) | ||
.via(flowFactory()) | ||
.runWith(sinkIn.sink)(subFusingMaterializer) | ||
|
||
if (isAvailable(out)) { | ||
sinkIn.pull() | ||
} | ||
|
@@ -419,26 +431,35 @@ private abstract class RestartWithBackoffLogic[S <: Shape]( | |
maxRestarts: Int) extends TimerGraphStageLogicWithLogging(shape) { | ||
var restartCount = 0 | ||
var resetDeadline = minBackoff.fromNow | ||
|
||
// This is effectively only used for flows, if either the main inlet or outlet of this stage finishes, then we | ||
// don't want to restart the sub inlet when it finishes, we just finish normally. | ||
var finishing = false | ||
|
||
protected def startGraph(): Unit | ||
protected def backoff(): Unit | ||
|
||
/** | ||
* @param out The permanent outlet | ||
* @return A sub sink inlet that's sink is attached to the wrapped stage | ||
*/ | ||
protected final def createSubInlet[T](out: Outlet[T]): SubSinkInlet[T] = { | ||
val sinkIn = new SubSinkInlet[T](s"RestartWithBackoff$name.subIn") | ||
|
||
sinkIn.setHandler(new InHandler { | ||
override def onPush() = push(out, sinkIn.grab()) | ||
|
||
override def onUpstreamFinish() = { | ||
if (finishing || maxRestartsReached() || onlyOnFailures) { | ||
complete(out) | ||
} else { | ||
log.debug("Restarting graph due to finished upstream") | ||
scheduleRestartTimer() | ||
} | ||
} | ||
|
||
/* | ||
* Upstream in this context is the wrapped stage. | ||
*/ | ||
override def onUpstreamFailure(ex: Throwable) = { | ||
if (finishing || maxRestartsReached()) { | ||
fail(out, ex) | ||
|
@@ -456,10 +477,13 @@ private abstract class RestartWithBackoffLogic[S <: Shape]( | |
sinkIn.cancel() | ||
} | ||
}) | ||
|
||
sinkIn | ||
} | ||
|
||
/** | ||
* @param in The permanent inlet for this stage | ||
* @return Temporary SubSourceOutlet for this "restart" | ||
*/ | ||
protected final def createSubOutlet[T](in: Inlet[T]): SubSourceOutlet[T] = { | ||
val sourceOut = new SubSourceOutlet[T](s"RestartWithBackoff$name.subOut") | ||
|
||
|
@@ -471,11 +495,17 @@ private abstract class RestartWithBackoffLogic[S <: Shape]( | |
pull(in) | ||
} | ||
} | ||
|
||
/* | ||
* Downstream in this context is the wrapped stage. | ||
* | ||
* Can either be a failure or a cancel in the wrapped state. | ||
* onlyOnFailures is thus racy so a delay to cancellation is added in the case of a flow. | ||
*/ | ||
override def onDownstreamFinish() = { | ||
if (finishing || maxRestartsReached() || onlyOnFailures) { | ||
cancel(in) | ||
} else { | ||
log.debug("Graph in finished") | ||
scheduleRestartTimer() | ||
} | ||
} | ||
|
@@ -498,7 +528,7 @@ private abstract class RestartWithBackoffLogic[S <: Shape]( | |
sourceOut | ||
} | ||
|
||
protected final def maxRestartsReached() = { | ||
protected final def maxRestartsReached(): Boolean = { | ||
// Check if the last start attempt was more than the minimum backoff | ||
if (resetDeadline.isOverdue()) { | ||
log.debug("Last restart attempt was more than {} ago, resetting restart count", minBackoff) | ||
|
@@ -508,7 +538,7 @@ private abstract class RestartWithBackoffLogic[S <: Shape]( | |
} | ||
|
||
// Set a timer to restart after the calculated delay | ||
protected final def scheduleRestartTimer() = { | ||
protected final def scheduleRestartTimer(): Unit = { | ||
val restartDelay = BackoffSupervisor.calculateDelay(restartCount, minBackoff, maxBackoff, randomFactor) | ||
log.debug("Restarting graph in {}", restartDelay) | ||
scheduleOnce("RestartTimer", restartDelay) | ||
|
@@ -526,3 +556,51 @@ private abstract class RestartWithBackoffLogic[S <: Shape]( | |
// When the stage starts, start the source | ||
override def preStart() = startGraph() | ||
} | ||
|
||
object RestartWithBackoffFlow { | ||
|
||
/** | ||
* Temporary attribute that can override the time a [[RestartWithBackoffFlow]] waits | ||
* for a failure before cancelling. | ||
* | ||
* See https://github.com/akka/akka/issues/24529 | ||
* | ||
* Will be removed if/when cancellation can include a cause. | ||
*/ | ||
@ApiMayChange | ||
case class Delay(duration: FiniteDuration) extends Attribute | ||
|
||
/** | ||
* Returns a flow that is almost identity but delays propagation of cancellation from downstream to upstream. | ||
* | ||
* Once the down stream is finish calls to onPush are ignored. | ||
*/ | ||
private def delayCancellation[T](duration: FiniteDuration): Flow[T, T, NotUsed] = | ||
Flow.fromGraph(new DelayCancellationStage(duration)) | ||
|
||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
private final class DelayCancellationStage[T](delay: FiniteDuration) extends SimpleLinearGraphStage[T] { | ||
|
||
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new TimerGraphStageLogic(shape) with InHandler with OutHandler with StageLogging { | ||
|
||
setHandlers(in, out, this) | ||
|
||
def onPush(): Unit = push(out, grab(in)) | ||
def onPull(): Unit = pull(in) | ||
|
||
override def onDownstreamFinish(): Unit = { | ||
scheduleOnce("CompleteState", delay) | ||
setHandler( | ||
in, | ||
new InHandler { | ||
def onPush(): Unit = {} | ||
} | ||
) | ||
} | ||
|
||
override protected def onTimer(timerKey: Any): Unit = { | ||
log.debug(s"Stage was canceled after delay of $delay") | ||
completeStage() | ||
} | ||
} | ||
} | ||
} |
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
So here, the attributes will not be passed on from the
RestartWithBackoffLogic
stage, unless there is something I don't know about thesubFusingMaterializer
.I think you'd expect that
restart(inner flow).withAttributes(Greatness(max))
would provide that attribute as a default for the inner flow when materialized as well. Could be that is missing currently.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes i'd expect that, you can put them on the inner flow but not on the restart bit for the inner flow