You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "Roiocam (via GitHub)" <gi...@apache.org> on 2023/09/21 16:01:09 UTC
[GitHub] [incubator-pekko] Roiocam opened a new pull request, #670: fix: keep stashBuffer when exception throw again #669
Roiocam opened a new pull request, #670:
URL: https://github.com/apache/incubator-pekko/pull/670
References #669
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1333337055
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1161,19 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "not msg lost on restart multiple times" in new FailingRestartBackoffTestSetup(cntValue = 5) {
+ val exception = new RuntimeException("mockException")
+ val msg: Throw = Throw(exception)
+ val ref = spawn(behv)
+ for (_ <- 1 to cntValue) {
+ ref ! msg
+ }
+ latch.await(1000, TimeUnit.MILLISECONDS)
+ for (_ <- 1 to cntValue) {
+ probe.expectMessage(msg)
+ }
Review Comment:
this test doesn't test the behavior.
I would expect sending N messages, to it and stash, and then send a `Throw` command, it throws and restart, when unstashing, the messages is processed successfully and not lost.
And another test is needed to test if it throw when unstashing, the stash will not be losted too.
This Pr fixed two issues, so better two tests?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam commented on pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#issuecomment-1741670149
@He-Pin I am sorry for that closed this PR for no reason, this PR is not only my contribution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1335746787
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -294,6 +297,44 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ class FailingConstructorTestSetupWithChild(failCount: Int, slowStopCount: Int) {
+ val failCounter = new AtomicInteger(0)
+ val probe = TestProbe[Event]("evt")
+ val slowStop = new CountDownLatch(slowStopCount)
+
+ class FailingConstructor(context: ActorContext[Command], monitor: ActorRef[Event])
+ extends AbstractBehavior[Command](context) {
+ monitor ! Started
+ context.spawn(targetBehavior(probe.ref, slowStop = Some(slowStop)), nextName())
+ if (failCounter.getAndIncrement() < failCount) {
+ throw TestException("simulated exc from constructor")
+ }
+
+ override def onMessage(message: Command): Behavior[Command] = {
+ monitor ! Pong(0)
Review Comment:
maybe better with a `Pong(counter++)`, it always reply `Pong(0)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1335075827
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -294,6 +297,44 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ class FailingConstructorTestSetupWithChild(failCount: Int, slowStopCount: Int) {
+ val failCounter = new AtomicInteger(0)
+ val probe = TestProbe[Event]("evt")
+ val slowStop = new CountDownLatch(slowStopCount)
+
+ class FailingConstructor(context: ActorContext[Command], monitor: ActorRef[Event])
+ extends AbstractBehavior[Command](context) {
+ monitor ! Started
+ context.spawn(targetBehavior(probe.ref, slowStop = Some(slowStop)), nextName())
+ if (failCounter.getAndIncrement() < failCount) {
+ throw TestException("simulated exc from constructor")
+ }
+
+ override def onMessage(message: Command): Behavior[Command] = {
+ monitor ! Pong(0)
+ Behaviors.same
+ }
+ }
+
+ def testMessageRetentionWhenStartException(strategy: SupervisorStrategy): Unit = {
+ val behv = supervise(setup[Command](ctx => new FailingConstructor(ctx, probe.ref)))
+ .onFailure[Exception](strategy)
+ val ref = spawn(behv)
+ probe.expectMessage(Started)
+ ref ! Ping(1)
+ ref ! Ping(2)
+ // unlock restart
+ slowStop.countDown()
Review Comment:
`RestarSupervisorStrategy`can only wait through its child terminated in order for messages to be stashed, otherwise the restart will immediately succeed.
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1174,55 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "ensure unhandled message retention during multiple exception when restart" in {
+ testMessageRetentionWhenMultipleException(SupervisorStrategy.restart.withStashCapacity(4))
+ }
+
+ "ensure unhandled message retention during multiple exception when backoff" in {
+ testMessageRetentionWhenMultipleException(SupervisorStrategy.restartWithBackoff(10.millis, 10.millis,
+ 0).withStashCapacity(4))
+ }
+
+ def testMessageRetentionWhenMultipleException(strategy: SupervisorStrategy): Unit = {
+ val probe = TestProbe[Event]("evt")
+ val slowRestart = new CountDownLatch(1)
+ val behv =
+ Behaviors.supervise(targetBehavior(probe.ref, slowRestart = Some(slowRestart))).onFailure[Exc1](strategy)
+ val ref = spawn(behv)
+
+ // restart strategy require a latch in order to afford the opportunity to stash messages
+ val childProbe = TestProbe[Event]("childEvt")
+ val childSlowStop = new CountDownLatch(1)
+ val childName = nextName()
+ ref ! CreateChild(targetBehavior(childProbe.ref, slowStop = Some(childSlowStop)), childName)
+ ref ! GetState
+ probe.expectMessageType[State].children.keySet should ===(Set(childName))
+
+ ref ! Throw(new Exc1)
+ ref ! Throw(new Exc1)
Review Comment:
This case simulates an exception occurring during `stashBuffer.unstashAll` and verifies that 'restartingInProgress` will not be incorrectly reset.
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1174,55 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "ensure unhandled message retention during multiple exception when restart" in {
+ testMessageRetentionWhenMultipleException(SupervisorStrategy.restart.withStashCapacity(4))
+ }
+
+ "ensure unhandled message retention during multiple exception when backoff" in {
+ testMessageRetentionWhenMultipleException(SupervisorStrategy.restartWithBackoff(10.millis, 10.millis,
+ 0).withStashCapacity(4))
+ }
+
+ def testMessageRetentionWhenMultipleException(strategy: SupervisorStrategy): Unit = {
+ val probe = TestProbe[Event]("evt")
+ val slowRestart = new CountDownLatch(1)
+ val behv =
+ Behaviors.supervise(targetBehavior(probe.ref, slowRestart = Some(slowRestart))).onFailure[Exc1](strategy)
+ val ref = spawn(behv)
+
+ // restart strategy require a latch in order to afford the opportunity to stash messages
+ val childProbe = TestProbe[Event]("childEvt")
+ val childSlowStop = new CountDownLatch(1)
+ val childName = nextName()
+ ref ! CreateChild(targetBehavior(childProbe.ref, slowStop = Some(childSlowStop)), childName)
+ ref ! GetState
+ probe.expectMessageType[State].children.keySet should ===(Set(childName))
+
+ ref ! Throw(new Exc1)
+ ref ! Throw(new Exc1)
+ ref ! Ping(1)
+ ref ! Ping(2)
+ // waiting for actor to restart, Pings will stashed
+ probe.expectNoMessage()
+ slowRestart.countDown()
+ probe.expectMessage(ReceivedSignal(PreRestart))
+ childSlowStop.countDown()
+ probe.expectMessage(ReceivedSignal(PreRestart))
+ probe.expectMessage(Pong(1))
+ probe.expectMessage(Pong(2))
+ }
+
+ "ensure stash message retention on start exception when restart" in new FailingConstructorTestSetupWithChild(
+ failCount = 2, slowStopCount = 1) {
+ testMessageRetentionWhenStartException(SupervisorStrategy.restart.withStashCapacity(4).withLimit(4, 10.seconds))
+ }
+
+ "ensure stash message retention on start exception when backoff" in new FailingConstructorTestSetupWithChild(
+ failCount = 2, slowStopCount = 1) {
+ testMessageRetentionWhenStartException(SupervisorStrategy.restartWithBackoff(10.millis, 10.millis,
Review Comment:
This case simulates the scenario where an Actor encounters an exception, enters the restarting process, and stashes messages within this process.
but encounters an exception (during the startup phase) before reaching `stashBuffer.unstashAll`. It verifies that `stashBuffer` will not be recreated in `prepareRestart` when the restart hasn't been completed.
It is correct behavior for `stashBuffer` to be repeatedly created When the exception does not occur repeatedly within a short period of time. At this moment, `stashBuffer` will also not have any unhandled messages.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1335051388
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1161,19 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "not msg lost on restart multiple times" in new FailingRestartBackoffTestSetup(cntValue = 5) {
+ val exception = new RuntimeException("mockException")
+ val msg: Throw = Throw(exception)
+ val ref = spawn(behv)
+ for (_ <- 1 to cntValue) {
+ ref ! msg
+ }
+ latch.await(1000, TimeUnit.MILLISECONDS)
+ for (_ <- 1 to cntValue) {
+ probe.expectMessage(msg)
+ }
Review Comment:
It is that two-issue mean `Restart` and `RestartBackoff` drop an unhandled message,
or "keep `restartingInProgress` before all messages handled" and "refill buffer on preperRestart".
I am struggling with this, and there is another case where, although retaining "restartingInProgress," buffer messages are still lost when encountering an "UnstashException" and the number of retries exceeds`maxRestart`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam closed pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam closed pull request #670: fix: keep stashBuffer when exception throw again #669
URL: https://github.com/apache/incubator-pekko/pull/670
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1335746787
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -294,6 +297,44 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ class FailingConstructorTestSetupWithChild(failCount: Int, slowStopCount: Int) {
+ val failCounter = new AtomicInteger(0)
+ val probe = TestProbe[Event]("evt")
+ val slowStop = new CountDownLatch(slowStopCount)
+
+ class FailingConstructor(context: ActorContext[Command], monitor: ActorRef[Event])
+ extends AbstractBehavior[Command](context) {
+ monitor ! Started
+ context.spawn(targetBehavior(probe.ref, slowStop = Some(slowStop)), nextName())
+ if (failCounter.getAndIncrement() < failCount) {
+ throw TestException("simulated exc from constructor")
+ }
+
+ override def onMessage(message: Command): Behavior[Command] = {
+ monitor ! Pong(0)
Review Comment:
maybe better with a `Pong(ping.seqNr)`, it always reply `Pong(0)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1335051780
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1161,19 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "not msg lost on restart multiple times" in new FailingRestartBackoffTestSetup(cntValue = 5) {
+ val exception = new RuntimeException("mockException")
+ val msg: Throw = Throw(exception)
+ val ref = spawn(behv)
+ for (_ <- 1 to cntValue) {
+ ref ! msg
+ }
+ latch.await(1000, TimeUnit.MILLISECONDS)
+ for (_ <- 1 to cntValue) {
+ probe.expectMessage(msg)
+ }
Review Comment:
When the restart time exceeded, drop the buffer seems fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1333447377
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1161,19 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "not msg lost on restart multiple times" in new FailingRestartBackoffTestSetup(cntValue = 5) {
+ val exception = new RuntimeException("mockException")
+ val msg: Throw = Throw(exception)
+ val ref = spawn(behv)
+ for (_ <- 1 to cntValue) {
+ ref ! msg
+ }
+ latch.await(1000, TimeUnit.MILLISECONDS)
+ for (_ <- 1 to cntValue) {
+ probe.expectMessage(msg)
+ }
Review Comment:
Yes, that's what I learnt from the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1333337055
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1161,19 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "not msg lost on restart multiple times" in new FailingRestartBackoffTestSetup(cntValue = 5) {
+ val exception = new RuntimeException("mockException")
+ val msg: Throw = Throw(exception)
+ val ref = spawn(behv)
+ for (_ <- 1 to cntValue) {
+ ref ! msg
+ }
+ latch.await(1000, TimeUnit.MILLISECONDS)
+ for (_ <- 1 to cntValue) {
+ probe.expectMessage(msg)
+ }
Review Comment:
this test doesn't test the behavior.
I would expect sending N messages, to it and stash, and then send a `Throw` command, it throws and restart, when unstashing, the message is processed successfully.
And another test is needed to test if it throw when unstashing, the stash will not be losted too.
This Pr fixed two issues, so better two tests?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1333337055
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1161,19 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "not msg lost on restart multiple times" in new FailingRestartBackoffTestSetup(cntValue = 5) {
+ val exception = new RuntimeException("mockException")
+ val msg: Throw = Throw(exception)
+ val ref = spawn(behv)
+ for (_ <- 1 to cntValue) {
+ ref ! msg
+ }
+ latch.await(1000, TimeUnit.MILLISECONDS)
+ for (_ <- 1 to cntValue) {
+ probe.expectMessage(msg)
+ }
Review Comment:
this test doesn't test the behavior
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: keep stashBuffer when exception throw again #669 [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1336039225
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -294,6 +297,44 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ class FailingConstructorTestSetupWithChild(failCount: Int, slowStopCount: Int) {
+ val failCounter = new AtomicInteger(0)
+ val probe = TestProbe[Event]("evt")
+ val slowStop = new CountDownLatch(slowStopCount)
+
+ class FailingConstructor(context: ActorContext[Command], monitor: ActorRef[Event])
+ extends AbstractBehavior[Command](context) {
+ monitor ! Started
+ context.spawn(targetBehavior(probe.ref, slowStop = Some(slowStop)), nextName())
+ if (failCounter.getAndIncrement() < failCount) {
+ throw TestException("simulated exc from constructor")
+ }
+
+ override def onMessage(message: Command): Behavior[Command] = {
+ monitor ! Pong(0)
Review Comment:
Makes sense, In this way, we can gain the guarantee of message ordering.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1333334633
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1161,19 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "not msg lost on restart multiple times" in new FailingRestartBackoffTestSetup(cntValue = 5) {
Review Comment:
Doesn't lost stashBuffer...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1333352297
##########
actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/Supervision.scala:
##########
@@ -398,8 +404,10 @@ private class RestartSupervisor[T, Thr <: Throwable: ClassTag](initial: Behavior
val newBehavior = Behavior.validateAsInitial(Behavior.start(initial, ctx.asInstanceOf[TypedActorContext[T]]))
val nextBehavior = restartingInProgress match {
case OptionVal.Some((stashBuffer, _)) =>
+ val behavior = stashBuffer.unstashAll(newBehavior.unsafeCast)
+ // reset stash state if not more exception throw
restartingInProgress = OptionVal.None
- stashBuffer.unstashAll(newBehavior.unsafeCast)
+ behavior
Review Comment:
Yes, because unhandled messages are stored in the StashBuffer within the `restartingInProgress` field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1333425038
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -1133,6 +1161,19 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ "not msg lost on restart multiple times" in new FailingRestartBackoffTestSetup(cntValue = 5) {
+ val exception = new RuntimeException("mockException")
+ val msg: Throw = Throw(exception)
+ val ref = spawn(behv)
+ for (_ <- 1 to cntValue) {
+ ref ! msg
+ }
+ latch.await(1000, TimeUnit.MILLISECONDS)
+ for (_ <- 1 to cntValue) {
+ probe.expectMessage(msg)
+ }
Review Comment:
The "StashBuffer" is utilized only during the restart phase and can be cleared once the restart is successful and all messages have been processed. This PR essentially addresses the following issue:
- Reusing the existing "StashBuffer" object during the preparation of a restart to prevent unhandled message loss.
- Resetting the "restartingInProgress" object only after all unhandled messages have been processed following the completion of the restart, thus avoiding the loss of the "StashBuffer".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1333331567
##########
actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/Supervision.scala:
##########
@@ -398,8 +404,10 @@ private class RestartSupervisor[T, Thr <: Throwable: ClassTag](initial: Behavior
val newBehavior = Behavior.validateAsInitial(Behavior.start(initial, ctx.asInstanceOf[TypedActorContext[T]]))
val nextBehavior = restartingInProgress match {
case OptionVal.Some((stashBuffer, _)) =>
+ val behavior = stashBuffer.unstashAll(newBehavior.unsafeCast)
+ // reset stash state if not more exception throw
restartingInProgress = OptionVal.None
- stashBuffer.unstashAll(newBehavior.unsafeCast)
+ behavior
Review Comment:
Only reset if the unstashAll processed success, otherwise the unprocessed messages will be lost right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1336029965
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -294,6 +297,44 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ class FailingConstructorTestSetupWithChild(failCount: Int, slowStopCount: Int) {
+ val failCounter = new AtomicInteger(0)
+ val probe = TestProbe[Event]("evt")
+ val slowStop = new CountDownLatch(slowStopCount)
+
+ class FailingConstructor(context: ActorContext[Command], monitor: ActorRef[Event])
+ extends AbstractBehavior[Command](context) {
+ monitor ! Started
+ context.spawn(targetBehavior(probe.ref, slowStop = Some(slowStop)), nextName())
+ if (failCounter.getAndIncrement() < failCount) {
+ throw TestException("simulated exc from constructor")
+ }
+
+ override def onMessage(message: Command): Behavior[Command] = {
+ monitor ! Pong(0)
Review Comment:
but then we can not gain the guarantee of message ordering.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] Roiocam commented on a diff in pull request #670: fix: keep stashBuffer when exception throw again #669
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #670:
URL: https://github.com/apache/incubator-pekko/pull/670#discussion_r1336029965
##########
actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala:
##########
@@ -294,6 +297,44 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
+ class FailingConstructorTestSetupWithChild(failCount: Int, slowStopCount: Int) {
+ val failCounter = new AtomicInteger(0)
+ val probe = TestProbe[Event]("evt")
+ val slowStop = new CountDownLatch(slowStopCount)
+
+ class FailingConstructor(context: ActorContext[Command], monitor: ActorRef[Event])
+ extends AbstractBehavior[Command](context) {
+ monitor ! Started
+ context.spawn(targetBehavior(probe.ref, slowStop = Some(slowStop)), nextName())
+ if (failCounter.getAndIncrement() < failCount) {
+ throw TestException("simulated exc from constructor")
+ }
+
+ override def onMessage(message: Command): Behavior[Command] = {
+ monitor ! Pong(0)
Review Comment:
but then we can not gain the guarantee of message ordering.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org