feat: ChangeEvent from Durable State by patriknw · Pull Request #32254 · akka/akka-core

patriknw · 2023-12-08T12:35:11Z

by storing additional change events when Durable State is updated and deleted we make it possible to use all nice event sourced Projection capabilities with Durable State, including Projections over gRPC
otherwise we would have to duplicate many many things to use the existing DurableStateChange and the queries based on that
we might even deprecate the existing DurableStateChange if this works out, but that can be a later decision

TODO:

docs

* by storing additional change events when Durable State is updated and deleted we make it possible to use all nice event sourced Projection capabilities with Durable State, including Projections over gRPC * otherwise we would have to duplicate many many things to use the existing DurableStateChange and the queries based on that * we might even deprecate the existing DurableStateChange if this works out, but that can be a later decision

patriknw · 2023-12-08T12:38:49Z

akka-persistence-typed/src/main/scala/akka/persistence/typed/state/internal/Running.scala

        behavior

-      case callback: Callback[_] =>
+      case callback: Callback[Any] @unchecked =>


This is a weird bug that we haven't noticed before. In the new test this fails when using:

Effect.delete().thenReply(replyTo)((_: String) => Done)

No compilation errors, but actually some red squiggles in IntelliJ. The runtime error was:

java.lang.ClassCastException: class java.lang.String cannot be cast to class scala.runtime.Nothing$ (java.lang.String is in module java.base of loader 'bootstrap'; scala.runtime.Nothing$ is in unnamed module of loader 'app') at akka.persistence.typed.state.internal.ReplyEffectImpl$$anonfun$$lessinit$greater$1.apply(SideEffect.scala:32) at akka.persistence.typed.state.internal.ReplyEffectImpl$$anonfun$$lessinit$greater$1.apply(SideEffect.scala:32) at akka.persistence.typed.state.internal.Running.applySideEffect(Running.scala:381) at akka.persistence.typed.state.internal.Running.applySideEffects(Running.scala:359) at akka.persistence.typed.state.internal.Running$PersistingState.onDeleteSuccess(Running.scala:283) at akka.persistence.typed.state.internal.Running$PersistingState.onMessage(Running.scala:236) at akka.persistence.typed.state.internal.Running$PersistingState.onMessage(Running.scala:222) at akka.actor.typed.scaladsl.AbstractBehavior.receive(AbstractBehavior.scala:84) at akka.actor.typed.Behavior$.interpret(Behavior.scala:282) at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:238) at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:57) at akka.persistence.typed.state.internal.DurableStateBehaviorImpl$$anon$1.aroundReceive(DurableStateBehaviorImpl.scala:144) at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:85) at akka.actor.typed.Behavior$.interpret(Behavior.scala:282) at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:238) at akka.actor.typed.internal.InterceptorImpl$$anon$2.apply(InterceptorImpl.scala:57) at akka.actor.typed.internal.SimpleSupervisor.aroundReceive(Supervision.scala:132) at akka.actor.typed.internal.InterceptorImpl.receive(InterceptorImpl.scala:85) at akka.actor.typed.Behavior$.interpret(Behavior.scala:282) at akka.actor.typed.Behavior$.interpretMessage(Behavior.scala:238) at akka.actor.typed.internal.adapter.ActorAdapter.handleMessage(ActorAdapter.scala:133) at akka.actor.typed.internal.adapter.ActorAdapter.$anonfun$aroundReceive$2(ActorAdapter.scala:101) at akka.actor.typed.internal.adapter.ActorAdapter.$anonfun$aroundReceive$2$adapted(ActorAdapter.scala:97) at akka.actor.typed.internal.adapter.ActorAdapter.withSafelyAdapted(ActorAdapter.scala:204) at akka.actor.typed.internal.adapter.ActorAdapter.aroundReceive(ActorAdapter.scala:97) at akka.actor.ActorCell.receiveMessage(ActorCell.scala:579) at akka.actor.ActorCell.invoke(ActorCell.scala:547) at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270) at akka.dispatch.Mailbox.run(Mailbox.scala:231) at akka.dispatch.Mailbox.exec(Mailbox.scala:243) at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373) at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182) at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655) at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622) at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165)

Same fix in EventSourcedBehavior.

...nce-typed/src/test/java/jdocs/akka/persistence/typed/DurableStatePersistentBehaviorTest.java

...istence-typed/src/main/scala/akka/persistence/typed/state/javadsl/DurableStateBehavior.scala

patriknw · 2023-12-12T07:52:58Z

This is ready for final review. Regarding reference docs, I added plugin api sections as we had for the state store, but I will add usage docs of the change event handler in Akka Projections docs since that is the purpose.

johanandren

LGTM!

johanandren · 2023-12-11T17:07:28Z

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/Running.scala

        behavior

-      case callback: Callback[_] =>
+      case callback: Callback[Any] @unchecked =>


Unchecked any is the best any.

johanandren · 2023-12-11T17:07:55Z

akka-persistence-typed/src/main/scala/akka/persistence/typed/state/internal/BehaviorSetup.scala

+  if (changeEventHandler.isDefined && !durableStateStore.isInstanceOf[DurableStateUpdateWithChangeEventStore[_]])
+    new IllegalArgumentException(
+      "Change event handler was defined but the DurableStateStore " +
+      s"[${durableStateStore.getClass.getName}] doesn't implement [DurableStateUpdateWithChangeEventStore]")


patriknw · 2023-12-12T09:21:40Z

...rsistence-typed/src/main/scala/akka/persistence/typed/state/javadsl/ChangeEventHandler.scala

+ * The `updateHandler` and `deleteHandler` are invoked after the ordinary command handler. Be aware of that
+ * if the state is mutable and modified by the command handler the previous state parameter of the `updateHandler`
+ * will also include the modification, since it's the same instance. If that is problem you need to use
+ * immutable state and create a new state instance when modifying it in the command handler.


added this caveat about mutable state

...rsistence-typed/src/main/scala/akka/persistence/typed/state/javadsl/ChangeEventHandler.scala

johanandren · 2023-12-12T09:42:12Z

...rsistence-typed/src/main/scala/akka/persistence/typed/state/javadsl/ChangeEventHandler.scala

+ * The `updateHandler` and `deleteHandler` are invoked after the ordinary command handler. Be aware of that
+ * if the state is mutable and modified by the command handler the previous state parameter of the `updateHandler`
+ * will also include the modification, since it's the same instance. If that is problem you need to use
+ * immutable state and create a new state instance when modifying it in the command handler.


...sistence-typed/src/main/scala/akka/persistence/typed/state/scaladsl/ChangeEventHandler.scala

Co-authored-by: Johan Andrén <johan@markatta.com>

* by storing additional change events when Durable State is updated and deleted we make it possible to use all nice event sourced Projection capabilities with Durable State, including Projections over gRPC * otherwise we would have to duplicate many many things to use the existing DurableStateChange and the queries based on that * we might even deprecate the existing DurableStateChange if this works out, but that can be a later decision * ApiMayChange * reference docs for plugin api * note about mutable state

patriknw commented Dec 8, 2023

View reviewed changes

...nce-typed/src/test/java/jdocs/akka/persistence/typed/DurableStatePersistentBehaviorTest.java Outdated Show resolved Hide resolved

patriknw added 2 commits December 11, 2023 10:55

add command parameter

1393b82

Scala 3 exhaustive match

7e3d123

patriknw force-pushed the wip-chg-evt-patriknw branch from dd4dc76 to 22d5bf5 Compare December 11, 2023 11:28

patriknw commented Dec 11, 2023

View reviewed changes

...istence-typed/src/main/scala/akka/persistence/typed/state/javadsl/DurableStateBehavior.scala Outdated Show resolved Hide resolved

change handler override in javadsl

ca4a7ad

patriknw force-pushed the wip-chg-evt-patriknw branch from 22d5bf5 to ca4a7ad Compare December 11, 2023 14:02

patriknw added 2 commits December 12, 2023 08:29

ApiMayChange

312ebee

reference docs for plugin api

fde4109

johanandren approved these changes Dec 12, 2023

View reviewed changes

note about mutable state

4eb6530

patriknw commented Dec 12, 2023

View reviewed changes

johanandren approved these changes Dec 12, 2023

View reviewed changes

patriknw and others added 2 commits December 12, 2023 10:52

Apply suggestions from code review

b3e714e

Co-authored-by: Johan Andrén <johan@markatta.com>

rename methods in javadsl ChangeEventHandler

9fe1313

johanandren approved these changes Dec 14, 2023

View reviewed changes

patriknw merged commit e71e1c5 into main Dec 14, 2023

patriknw deleted the wip-chg-evt-patriknw branch December 14, 2023 16:00

patriknw added this to the 2.9.1 milestone Dec 14, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Comments

feat: ChangeEvent from Durable State#32254

feat: ChangeEvent from Durable State#32254
patriknw merged 9 commits intomainfrom
wip-chg-evt-patriknw

patriknw commented Dec 8, 2023 •

edited

Loading

Uh oh!

patriknw Dec 8, 2023

Uh oh!

Uh oh!

Uh oh!

patriknw commented Dec 12, 2023

Uh oh!

johanandren left a comment

Uh oh!

johanandren Dec 11, 2023

Uh oh!

johanandren Dec 11, 2023

Uh oh!

patriknw Dec 12, 2023

Uh oh!

johanandren Dec 12, 2023

Uh oh!

Uh oh!

johanandren Dec 12, 2023

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Comments

Conversation

patriknw commented Dec 8, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

patriknw Dec 8, 2023

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

patriknw commented Dec 12, 2023

Uh oh!

johanandren left a comment

Choose a reason for hiding this comment

Uh oh!

johanandren Dec 11, 2023

Choose a reason for hiding this comment

Uh oh!

johanandren Dec 11, 2023

Choose a reason for hiding this comment

Uh oh!

patriknw Dec 12, 2023

Choose a reason for hiding this comment

Uh oh!

johanandren Dec 12, 2023

Choose a reason for hiding this comment

Uh oh!

Uh oh!

johanandren Dec 12, 2023

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

patriknw commented Dec 8, 2023 •

edited

Loading