You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "pjfanning (via GitHub)" <gi...@apache.org> on 2025/10/25 12:12:43 UTC
[PR] add tests for FlowStatefulMap [pekko]
pjfanning opened a new pull request, #2365:
URL: https://github.com/apache/pekko/pull/2365
* based on https://github.com/akka/akka-core/pull/31630
* Akka 2.7.0 changes are now available under Apache license
* this test fail in Pekko so it looks like we need changes to get them to work
--
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 close in cancel for statefulMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #2365:
URL: https://github.com/apache/pekko/pull/2365#discussion_r2462857655
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowStatefulMapSpec.scala:
##########
@@ -455,4 +454,86 @@ class FlowStatefulMapSpec extends StreamSpec {
.expectComplete()
}
+ "will not call onComplete twice on cancel when `onComplete` fails" in {
+ val closedCounter = new AtomicInteger(0)
+ val (source, sink) = TestSource()
+ .viaMat(Flow[Int].statefulMap(() => 23)((s, elem) => (s, elem),
+ _ => {
+ closedCounter.incrementAndGet()
+ throw TE("boom")
+ }))(Keep.left)
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ EventFilter[TE](occurrences = 1).intercept {
+ sink.request(1)
+ source.sendNext(1)
+ sink.expectNext(1)
+ sink.cancel()
+ source.expectCancellation()
+ }
+ closedCounter.get() should ===(1)
+ }
+
+ "emit onClose return value before restarting" in {
+ val stateCounter = new AtomicInteger(0)
+ val (source, sink) = TestSource[String]()
+ .viaMat(Flow[String].statefulMap(() => stateCounter.incrementAndGet())({ (s, elem) =>
+ if (elem == "boom") throw TE("boom")
+ else (s, elem + s.toString)
+ }, _ => Some("onClose")))(Keep.left)
+ .withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+ sink.request(1)
+ source.sendNext("one")
+ sink.expectNext("one1")
+ sink.request(1)
+ source.sendNext("boom")
+ sink.expectNext("onClose")
+ sink.request(1)
+ source.sendNext("two")
+ sink.expectNext("two2")
+ sink.cancel()
+ source.expectCancellation()
+ }
+
+ "not allow null state" in {
+ EventFilter[NullPointerException](occurrences = 1).intercept {
+ Source
+ .single("one")
+ .statefulMap(() => null: String)((s, t) => (s, t), _ => None)
+ .runWith(Sink.head)
+ .failed
+ .futureValue shouldBe a[NullPointerException]
+ }
+ }
+
+ "not allow null next state" in {
+ EventFilter[NullPointerException](occurrences = 1).intercept {
+ Source
+ .single("one")
+ .statefulMap(() => "state")((_, t) => (null, t), _ => None)
+ .runWith(Sink.seq)
+ .failed
+ .futureValue shouldBe a[NullPointerException]
+ }
+ }
+
+ "not allow null state on restart" in {
Review Comment:
in theory, we can change behaviour in 1.3.0 - but if we truly want to keep null support, we could try to modify this test and the next one and then update the code to make the modified tests work
--
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 close in cancel for statefulMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2365:
URL: https://github.com/apache/pekko/pull/2365#issuecomment-3447009237
I will take care of this @pjfanning
--
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 close in cancel for statefulMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning merged PR #2365:
URL: https://github.com/apache/pekko/pull/2365
--
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 close in cancel for statefulMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #2365:
URL: https://github.com/apache/pekko/pull/2365#discussion_r2462857655
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowStatefulMapSpec.scala:
##########
@@ -455,4 +454,86 @@ class FlowStatefulMapSpec extends StreamSpec {
.expectComplete()
}
+ "will not call onComplete twice on cancel when `onComplete` fails" in {
+ val closedCounter = new AtomicInteger(0)
+ val (source, sink) = TestSource()
+ .viaMat(Flow[Int].statefulMap(() => 23)((s, elem) => (s, elem),
+ _ => {
+ closedCounter.incrementAndGet()
+ throw TE("boom")
+ }))(Keep.left)
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ EventFilter[TE](occurrences = 1).intercept {
+ sink.request(1)
+ source.sendNext(1)
+ sink.expectNext(1)
+ sink.cancel()
+ source.expectCancellation()
+ }
+ closedCounter.get() should ===(1)
+ }
+
+ "emit onClose return value before restarting" in {
+ val stateCounter = new AtomicInteger(0)
+ val (source, sink) = TestSource[String]()
+ .viaMat(Flow[String].statefulMap(() => stateCounter.incrementAndGet())({ (s, elem) =>
+ if (elem == "boom") throw TE("boom")
+ else (s, elem + s.toString)
+ }, _ => Some("onClose")))(Keep.left)
+ .withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+ sink.request(1)
+ source.sendNext("one")
+ sink.expectNext("one1")
+ sink.request(1)
+ source.sendNext("boom")
+ sink.expectNext("onClose")
+ sink.request(1)
+ source.sendNext("two")
+ sink.expectNext("two2")
+ sink.cancel()
+ source.expectCancellation()
+ }
+
+ "not allow null state" in {
+ EventFilter[NullPointerException](occurrences = 1).intercept {
+ Source
+ .single("one")
+ .statefulMap(() => null: String)((s, t) => (s, t), _ => None)
+ .runWith(Sink.head)
+ .failed
+ .futureValue shouldBe a[NullPointerException]
+ }
+ }
+
+ "not allow null next state" in {
+ EventFilter[NullPointerException](occurrences = 1).intercept {
+ Source
+ .single("one")
+ .statefulMap(() => "state")((_, t) => (null, t), _ => None)
+ .runWith(Sink.seq)
+ .failed
+ .futureValue shouldBe a[NullPointerException]
+ }
+ }
+
+ "not allow null state on restart" in {
Review Comment:
in theory, we can change behaviour in 1.3.0 - but if we truly want to keep null support, we could try to modify this test and the other new null tests and then update the code to make the modified tests work
--
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 close in cancel for statefulMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #2365:
URL: https://github.com/apache/pekko/pull/2365#discussion_r2462879139
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowStatefulMapSpec.scala:
##########
@@ -455,4 +454,86 @@ class FlowStatefulMapSpec extends StreamSpec {
.expectComplete()
}
+ "will not call onComplete twice on cancel when `onComplete` fails" in {
+ val closedCounter = new AtomicInteger(0)
+ val (source, sink) = TestSource()
+ .viaMat(Flow[Int].statefulMap(() => 23)((s, elem) => (s, elem),
+ _ => {
+ closedCounter.incrementAndGet()
+ throw TE("boom")
+ }))(Keep.left)
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ EventFilter[TE](occurrences = 1).intercept {
+ sink.request(1)
+ source.sendNext(1)
+ sink.expectNext(1)
+ sink.cancel()
+ source.expectCancellation()
+ }
+ closedCounter.get() should ===(1)
+ }
+
+ "emit onClose return value before restarting" in {
+ val stateCounter = new AtomicInteger(0)
+ val (source, sink) = TestSource[String]()
+ .viaMat(Flow[String].statefulMap(() => stateCounter.incrementAndGet())({ (s, elem) =>
+ if (elem == "boom") throw TE("boom")
+ else (s, elem + s.toString)
+ }, _ => Some("onClose")))(Keep.left)
+ .withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+ sink.request(1)
+ source.sendNext("one")
+ sink.expectNext("one1")
+ sink.request(1)
+ source.sendNext("boom")
+ sink.expectNext("onClose")
+ sink.request(1)
+ source.sendNext("two")
+ sink.expectNext("two2")
+ sink.cancel()
+ source.expectCancellation()
+ }
+
+ "not allow null state" in {
+ EventFilter[NullPointerException](occurrences = 1).intercept {
+ Source
+ .single("one")
+ .statefulMap(() => null: String)((s, t) => (s, t), _ => None)
+ .runWith(Sink.head)
+ .failed
+ .futureValue shouldBe a[NullPointerException]
+ }
+ }
+
+ "not allow null next state" in {
+ EventFilter[NullPointerException](occurrences = 1).intercept {
+ Source
+ .single("one")
+ .statefulMap(() => "state")((_, t) => (null, t), _ => None)
+ .runWith(Sink.seq)
+ .failed
+ .futureValue shouldBe a[NullPointerException]
+ }
+ }
+
+ "not allow null state on restart" in {
Review Comment:
I still like to keep the old behavior, we can improve both document and test.
--
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 close in cancel for statefulMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #2365:
URL: https://github.com/apache/pekko/pull/2365#discussion_r2462850577
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala:
##########
@@ -2250,69 +2260,86 @@ private[pekko] final class StatefulMap[S, In, Out](create: () => S, f: (S, In) =
new GraphStageLogic(shape) with InHandler with OutHandler {
lazy val decider: Decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
- private var state: S = _
- private var needInvokeOnCompleteCallback: Boolean = false
+ private var state: OptionVal[S] = OptionVal.none
- override def preStart(): Unit = {
- state = create()
- needInvokeOnCompleteCallback = true
- }
+ override def preStart(): Unit =
+ createNewState()
- override def onPush(): Unit =
+ override def onPush(): Unit = {
try {
val elem = grab(in)
- val (newState, newElem) = f(state, elem)
- state = newState
+ val (newState, newElem) = f(state.get, elem)
+ state = OptionVal.Some(newState)
+ throwIfNoState()
push(out, newElem)
} catch {
- case NonFatal(ex) =>
+ case ex: NullStateException => throw ex // don't cover with supervision
+ case NonFatal(ex) =>
decider(ex) match {
case Supervision.Stop => closeStateAndFail(ex)
case Supervision.Resume => pull(in)
- case Supervision.Restart => resetStateAndPull()
+ case Supervision.Restart => restartState()
}
}
+ }
- override def onUpstreamFinish(): Unit = closeStateAndComplete()
+ override def onUpstreamFinish(): Unit = {
+ completeStateIfNeeded() match {
+ case Some(elem) => emit(out, elem, () => completeStage())
+ case None => completeStage()
+ }
+ }
override def onUpstreamFailure(ex: Throwable): Unit = closeStateAndFail(ex)
override def onDownstreamFinish(cause: Throwable): Unit = {
- needInvokeOnCompleteCallback = false
- onComplete(state)
+ completeStateIfNeeded()
super.onDownstreamFinish(cause)
}
- private def resetStateAndPull(): Unit = {
- needInvokeOnCompleteCallback = false
- onComplete(state)
- state = create()
- needInvokeOnCompleteCallback = true;
- pull(in)
+ private def createNewState(): Unit = {
+ state = OptionVal.Some(create())
+ throwIfNoState()
}
- private def closeStateAndComplete(): Unit = {
- needInvokeOnCompleteCallback = false
- onComplete(state) match {
- case Some(elem) => emit(out, elem, () => completeStage())
- case None => completeStage()
+ private def restartState(): Unit = {
+ completeStateIfNeeded() match {
+ case Some(elem) =>
+ push(out, elem)
+ createNewState()
+ case None =>
+ createNewState()
+ // should always happen here but for good measure
+ if (!hasBeenPulled(in)) pull(in)
}
}
- private def closeStateAndFail(ex: Throwable): Unit = {
- needInvokeOnCompleteCallback = false
- onComplete(state) match {
+ private def closeStateAndFail(ex: Throwable): Unit =
+ completeStateIfNeeded() match {
case Some(elem) => emit(out, elem, () => failStage(ex))
case None => failStage(ex)
}
+
+ private def completeStateIfNeeded(): Option[Out] = {
+ state match {
+ case OptionVal.Some(s) =>
+ state = OptionVal.none[S]
+ onComplete(s)
+ case _ => None
+ }
}
override def onPull(): Unit = pull(in)
override def postStop(): Unit = {
- if (needInvokeOnCompleteCallback) {
- onComplete(state)
- }
+ completeStateIfNeeded()
+ }
+
+ private def throwIfNoState(): Unit = {
Review Comment:
This seems to change the behavior. @pjfanning I have not tested it ,but , if that's true, we should revert some changes.
--
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 close in cancel for statefulMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2365:
URL: https://github.com/apache/pekko/pull/2365#issuecomment-3446760802
@pjfanning A different is , IIRC, statefulmap in pekko support `null` as state, but not in akka
--
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 close in cancel for statefulMap [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #2365:
URL: https://github.com/apache/pekko/pull/2365#discussion_r2462852114
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowStatefulMapSpec.scala:
##########
@@ -455,4 +454,86 @@ class FlowStatefulMapSpec extends StreamSpec {
.expectComplete()
}
+ "will not call onComplete twice on cancel when `onComplete` fails" in {
+ val closedCounter = new AtomicInteger(0)
+ val (source, sink) = TestSource()
+ .viaMat(Flow[Int].statefulMap(() => 23)((s, elem) => (s, elem),
+ _ => {
+ closedCounter.incrementAndGet()
+ throw TE("boom")
+ }))(Keep.left)
+ .toMat(TestSink[Int]())(Keep.both)
+ .run()
+
+ EventFilter[TE](occurrences = 1).intercept {
+ sink.request(1)
+ source.sendNext(1)
+ sink.expectNext(1)
+ sink.cancel()
+ source.expectCancellation()
+ }
+ closedCounter.get() should ===(1)
+ }
+
+ "emit onClose return value before restarting" in {
+ val stateCounter = new AtomicInteger(0)
+ val (source, sink) = TestSource[String]()
+ .viaMat(Flow[String].statefulMap(() => stateCounter.incrementAndGet())({ (s, elem) =>
+ if (elem == "boom") throw TE("boom")
+ else (s, elem + s.toString)
+ }, _ => Some("onClose")))(Keep.left)
+ .withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+ sink.request(1)
+ source.sendNext("one")
+ sink.expectNext("one1")
+ sink.request(1)
+ source.sendNext("boom")
+ sink.expectNext("onClose")
+ sink.request(1)
+ source.sendNext("two")
+ sink.expectNext("two2")
+ sink.cancel()
+ source.expectCancellation()
+ }
+
+ "not allow null state" in {
+ EventFilter[NullPointerException](occurrences = 1).intercept {
+ Source
+ .single("one")
+ .statefulMap(() => null: String)((s, t) => (s, t), _ => None)
+ .runWith(Sink.head)
+ .failed
+ .futureValue shouldBe a[NullPointerException]
+ }
+ }
+
+ "not allow null next state" in {
+ EventFilter[NullPointerException](occurrences = 1).intercept {
+ Source
+ .single("one")
+ .statefulMap(() => "state")((_, t) => (null, t), _ => None)
+ .runWith(Sink.seq)
+ .failed
+ .futureValue shouldBe a[NullPointerException]
+ }
+ }
+
+ "not allow null state on restart" in {
Review Comment:
behavior change.
--
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 close in cancel for statefulMap [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #2365:
URL: https://github.com/apache/pekko/pull/2365#issuecomment-3446906711
I had a look at modifying the new code to not throw the NullStateException when the state evaluated as null but the new tests end up with NoSuchElementExceptions instead.
I think the options are:
* revert this and just keep the existing code and agree that this change is not important
* keep this change and mandate the users who rely on 'null' state can modify their code to use something like scala.None or java.lang.Optional.empty() instead.
* a significant rewrite of the new code that somehow allows nulls but that allows the new tests to work - this code is not an area of expertise for me so if someone wants to go this route, can they provide a PR?
wdyt? @He-Pin @mdedetrich @raboof @jrudolph
--
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