gRPC as Replicated ES transport #757

johanandren · 2022-12-05T18:25:40Z

Depends on upstream Akka changes: akka/akka#31771

patriknw

looks like a good start

...c/main/scala/akka/projection/grpc/replication/scaladsl/ReplicatedEventSourcingOverGrpc.scala

project/Dependencies.scala

patriknw · 2022-12-21T13:21:45Z

...c/main/scala/akka/projection/grpc/replication/scaladsl/ReplicatedEventSourcingOverGrpc.scala

+              destinationReplicaId.persistenceId.id,
+              replicatedEventMetadata.version,
+              envelope.event)
+            val askResult = entityRef.ask[Done](


It will be tempting to increase the parallelism of the mapAsync, but then we must be careful. Messages can be lost from here to the destination entity and we accept higher sequence numbers and can't validate exact sequence number. Then the following is possible:

send e1, e2, e3 to the entity

e1 arrives all fine

e2 is lost

e3 arrives and is accepted since it has a higher sequence number

there is a timeout of the ack for e2 and the stream is restarted

e2 is sent again, arrives but is handled as s duplicate because it has a lower sequence number than already handled e3

Once again we would need something like the mapAsyncPartitioned.

Lightweight internal mapAsyncPartitioned operator added in ef8f50d

patriknw

reviewed this in more detail, looking good

patriknw · 2023-01-09T15:10:31Z

...ction-grpc/src/test/java/akka/projection/grpc/replication/javdsl/ReplicationCompileTest.java

+
+  public static void start(ActorSystem<?> system) {
+   Set<Replica> otherReplicas = new HashSet<>();
+   otherReplicas.add(Replica.apply(


add javadsl Replica.create

fixed in 9c64f9f

akka-projection-grpc/src/main/protobuf/akka/projection/grpc/event_producer.proto

patriknw · 2023-01-09T15:25:33Z

akka-projection-grpc/src/main/scala/akka/projection/grpc/replication/ReplicationSettings.scala

+      parallelUpdates,
+      ReplicationProjectionProviderAdapter.toScala(replicationProjectionProvider))(classTag)
+  }
+}


this contains rather essential api with mix of Java create and Scala apply. Would it be difficult to move to the javadsl and scaladsl packages?

I think it was initially fine but outgrew a shared class as more and more settings became obvious, I'll create separate APIs

Massive change to separate APIs: 9c64f9f

And now even more separate in 1e0cbba to support Entity => Entity transformations for entity config in both DSLs

patriknw · 2023-01-09T15:31:43Z

akka-projection-grpc/src/main/scala/akka/projection/grpc/replication/scaladsl/Replication.scala

+      replicatedBehaviorFactory: ReplicationContext => EventSourcedBehavior[Command, Event, State])(
+      implicit system: ActorSystem[_]): Replication[Command] = {
+    require(
+      system.settings.config.getString("akka.actor.provider") == "cluster",


might be better to check ExtendedActorSystem.provider since it's still possible to configure cluster with full class name of the provider

Done in 971235e

akka-projection-grpc/src/main/scala/akka/projection/grpc/replication/scaladsl/Replication.scala

docs/src/main/paradox/grpc-replicated-event-sourcing-transport.md

...ojection-grpc/src/it/scala/akka/projection/grpc/replication/ReplicationIntegrationSpec.scala

docs/src/main/paradox/grpc-replicated-event-sourcing-transport.md

...ojection-grpc/src/it/scala/akka/projection/grpc/replication/ReplicationIntegrationSpec.scala

akka-projection-grpc/src/main/scala/akka/projection/grpc/replication/scaladsl/Replication.scala

pvlugter

Catching up on this with a read-through. Looks like it's coming together really nicely 👍🏼

...ojection-grpc/src/main/scala/akka/projection/grpc/replication/internal/ReplicationImpl.scala

akka-projection-grpc/src/main/scala/akka/projection/grpc/replication/scaladsl/Replication.scala

docs/src/main/paradox/grpc-replicated-event-sourcing-transport.md

johanandren · 2023-01-19T06:55:57Z

...ojection-grpc/src/main/scala/akka/projection/grpc/replication/internal/ReplicationImpl.scala

+   * Important: Note that this does not publish the endpoint, additional steps are needed!
+   */
+  def grpcReplication[Command, Event, State](settings: ReplicationSettingsImpl[Command])(
+      replicatedBehaviorFactory: ReplicationContext => EventSourcedBehavior[Command, Event, State])(


One problem with this type of factory is that since it needs to return an event sourced behavior it can't be composed to access context and timers

Peter has a solution for that in #769

johanandren · 2023-01-19T14:05:14Z

Latest commit depends on Akka changes from akka/akka#31817 and r2dbc changes in akka/akka-persistence-r2dbc#348

...-projection-grpc/src/main/scala/akka/projection/grpc/consumer/scaladsl/GrpcReadJournal.scala

...ojection-grpc/src/main/scala/akka/projection/grpc/replication/internal/ReplicationImpl.scala

…ration spec

…ndpoints in one server

johanandren · 2023-01-23T14:36:28Z

Two missing parts are samples and ability to compose behavior with setup/timers, Peter has pretty much completed samples and took a shot at composing so let's add that on top after merging this PR.

pvlugter

LGTM. Didn't re-review everything from earlier in detail. But with all the testing that's been done, and follow-ups for the final touches—massaging APIs some more, composability, samples, and extending the docs—looks good to merge now.

...ojection-grpc/src/it/scala/akka/projection/grpc/replication/ReplicationIntegrationSpec.scala

akka-projection-grpc/src/main/scala/akka/projection/grpc/replication/javadsl/Replication.scala

...rojection-grpc/src/test/scala/akka/projection/grpc/replication/ReplicationSettingsSpec.scala

Co-authored-by: Peter Vlugter <59895+pvlugter@users.noreply.github.com>

patriknw

looking good, I'll also look at Peter's samples and the benchmark project

patriknw · 2023-01-24T11:19:23Z

...ojection-grpc/src/it/scala/akka/projection/grpc/replication/ReplicationIntegrationSpec.scala

+       akka.persistence.r2dbc {
+          query {
+            refresh-interval = 500 millis
+            # reducing this to have quicker test, triggers backtracking earlier


it's also the backtracking.window that triggers the backtracking switch. When half of the backtracking.window has elapsed it will switch to backtracking

...-projection-grpc/src/main/scala/akka/projection/grpc/internal/EventProducerServiceImpl.scala

akka-projection-grpc/src/main/scala/akka/projection/grpc/producer/javadsl/Transformation.scala

...ojection-grpc/src/main/scala/akka/projection/grpc/replication/internal/ReplicationImpl.scala

docs/src/main/paradox/grpc-replicated-event-sourcing-transport.md

...rojection-grpc/src/test/scala/akka/projection/grpc/replication/ReplicationSettingsSpec.scala

docs/src/main/paradox/grpc.md

Also: * Re-renamed the JavaSourceProviderAdapter since it is used in Akka Persistence R2DBC * Some missing APIs added

…t still not working

patriknw

LGTM, when the small things have been fixed I think we can merge this and iterate on the remaining in separate PRs

pvlugter

Good to have the integration test using the Java DSL.

All looks fine and working in the samples, rebased on latest changes.

Looks like comments have been addressed, or follow-ups added, so seems good to merge.

pvlugter · 2023-01-24T23:13:21Z

akka-projection-core/src/main/scala/akka/projection/internal/SourceProviderAdapter.scala


 /**
 * INTERNAL API: Adapter from javadsl.SourceProvider to scaladsl.SourceProvider
 */
-@InternalApi private[projection] class SourceProviderAdapter[Offset, Envelope](
+@InternalStableApi private[projection] class SourceProviderAdapter[Offset, Envelope](


Useful to have this back to SourceProviderAdapter. I had added it back and deprecated this, with the rename to JavaSourceProviderAdapter, so the java sample could run without a local r2dbc snapshot.

pvlugter · 2023-01-24T23:13:47Z

akka-projection-core/src/main/scala/akka/projection/internal/SourceProviderAdapter.scala

+/**
+ * INTERNAL API: Adapter from scaladsl.SourceProvider with BySlicesSourceProvider to javadsl.SourceProvider with BySlicesSourceProvider
+ */
+@InternalApi private[projection] class ScalaBySlicesSourceProviderAdapter[Offset, Envelope](


👍🏼 the other missing adaptation for Java DSL.

pvlugter · 2023-01-25T00:05:05Z

Triggered rerun for CI job and unrelated failure (#465)

pvlugter · 2023-01-25T02:25:35Z

Will merge now, and retarget the samples PR.

patriknw reviewed Dec 7, 2022

View reviewed changes

...c/main/scala/akka/projection/grpc/replication/scaladsl/ReplicatedEventSourcingOverGrpc.scala Outdated Show resolved Hide resolved

project/Dependencies.scala Outdated Show resolved Hide resolved

patriknw reviewed Dec 15, 2022

View reviewed changes

project/Dependencies.scala Outdated Show resolved Hide resolved

patriknw reviewed Dec 21, 2022

View reviewed changes

johanandren force-pushed the wip-grpc-replicated-es-transport branch 2 times, most recently from 32fe923 to 5bf9e15 Compare December 23, 2022 12:48

johanandren force-pushed the wip-grpc-replicated-es-transport branch 4 times, most recently from 268b069 to ef8f50d Compare January 4, 2023 15:22

patriknw reviewed Jan 9, 2023

View reviewed changes

patriknw reviewed Jan 10, 2023

View reviewed changes

akka-projection-grpc/src/main/scala/akka/projection/grpc/replication/scaladsl/Replication.scala Outdated Show resolved Hide resolved

johanandren force-pushed the wip-grpc-replicated-es-transport branch from 5d4dd5f to 34c361c Compare January 10, 2023 15:37

pvlugter reviewed Jan 11, 2023

View reviewed changes

johanandren commented Jan 19, 2023

View reviewed changes

patriknw reviewed Jan 19, 2023

View reviewed changes

johanandren force-pushed the wip-grpc-replicated-es-transport branch from 4f59a4b to 929b6a6 Compare January 19, 2023 15:34

johanandren added 13 commits January 20, 2023 17:25

WIP: gRPC as Replicated ES transport

684b5c1

headers

7868202

WIP - one replicated event passing through, next event fails in integ…

77f9058

…ration spec

Integration test now passing with latest Akka PR update

5f3a4dc

Filter events in producer, API allowing to compose multiple AP gRPC e…

4285309

…ndpoints in one server

bump Akka 2.8.0-M3 with needed RES changes

41a77d2

Formatting and mima

3efdb4c

Compiles on Scala 2.12, sketch of Java API

77aa571

missing header

fdc5b48

Smore more

0100797

Unbreak the projection grpc test depending on log entries

b4c3682

Possibly workable Java API

77b31ac

Wrong class name for serializer registration

d48a549

johanandren changed the title ~~WIP: gRPC as Replicated ES transport~~ gRPC as Replicated ES transport Jan 23, 2023

johanandren marked this pull request as ready for review January 23, 2023 12:44

pvlugter approved these changes Jan 23, 2023

View reviewed changes

Apply suggestions from code review

54507ac

Co-authored-by: Peter Vlugter <59895+pvlugter@users.noreply.github.com>

johanandren mentioned this pull request Jan 24, 2023

Convenience provider factory for gRPC transport akka/akka-persistence-r2dbc#351

Closed

patriknw reviewed Jan 24, 2023

View reviewed changes

johanandren added 4 commits January 24, 2023 15:33

Integration test for the Java DSL (failing becacuse missing metadata)

79a95e3

Also: * Re-renamed the JavaSourceProviderAdapter since it is used in Akka Persistence R2DBC * Some missing APIs added

Always passing meta got rid of the metadata exception for Java DSL bu…

196e75e

…t still not working

Test passing, all is good

e5f88d8

Scala 2.12 fix

78aff58

johanandren mentioned this pull request Jan 24, 2023

Flow based projection doesn't stop on stop message #777

Closed

johanandren added 3 commits January 24, 2023 17:23

Move transformer fromScala -> toJava

331ecba

Include env var for replica id in config sample

3aafe68

Dependency versions from build, missing akka-projection-r2dbc dependency

8f4e637

patriknw approved these changes Jan 24, 2023

View reviewed changes

johanandren added 5 commits January 24, 2023 17:43

new over apply

9c62df0

Mention no need for shared multi dc cluster

a56def8

Auto-filter self replica from otherReplicas

c898d1e

Some helpful startup logging

2355b5a

infoN

492b96d

pvlugter approved these changes Jan 24, 2023

View reviewed changes

pvlugter mentioned this pull request Jan 25, 2023

Add samples for replicated event sourcing over gRPC #769

Merged

pvlugter merged commit ee1d5e4 into main Jan 25, 2023

pvlugter deleted the wip-grpc-replicated-es-transport branch January 25, 2023 02:47

johanandren added this to the 1.4.0-M2 milestone Jan 25, 2023

This was referenced Jan 25, 2023

feat: Convenience factories for using with Akka Projection gRPC akka/akka-persistence-r2dbc#352

Merged

doc: Replicated event sourcing gRPC doc improvements #785

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

gRPC as Replicated ES transport #757

gRPC as Replicated ES transport #757

johanandren commented Dec 5, 2022 •

edited

patriknw left a comment

patriknw Dec 21, 2022

johanandren Jan 10, 2023

patriknw left a comment

patriknw Jan 9, 2023

johanandren Jan 10, 2023

patriknw Jan 9, 2023

johanandren Jan 10, 2023

johanandren Jan 10, 2023

johanandren Jan 24, 2023

patriknw Jan 9, 2023

johanandren Jan 10, 2023

pvlugter left a comment

johanandren Jan 19, 2023

johanandren Jan 24, 2023

johanandren commented Jan 19, 2023

johanandren commented Jan 23, 2023

pvlugter left a comment

patriknw left a comment

patriknw Jan 24, 2023

patriknw left a comment

pvlugter left a comment

pvlugter Jan 24, 2023

pvlugter Jan 24, 2023

pvlugter commented Jan 25, 2023

pvlugter commented Jan 25, 2023

gRPC as Replicated ES transport #757

gRPC as Replicated ES transport #757

Conversation

johanandren commented Dec 5, 2022 • edited

patriknw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patriknw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pvlugter left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johanandren commented Jan 19, 2023

johanandren commented Jan 23, 2023

pvlugter left a comment

Choose a reason for hiding this comment

patriknw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patriknw left a comment

Choose a reason for hiding this comment

pvlugter left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pvlugter commented Jan 25, 2023

pvlugter commented Jan 25, 2023

johanandren commented Dec 5, 2022 •

edited