Introduce Projections spi #2727

ignasi35 · 2020-02-21T14:39:43Z

No description provided.

ignasi35 · 2020-03-06T16:34:54Z

...la/com/lightbend/lagom/internal/javadsl/persistence/cassandra/CassandraReadSideHandler.scala

@@ -21,7 +21,7 @@ import com.lightbend.lagom.javadsl.persistence.ReadSideProcessor.ReadSideHandler
 import com.lightbend.lagom.javadsl.persistence.cassandra.CassandraSession
 import com.lightbend.lagom.javadsl.persistence.AggregateEvent
 import com.lightbend.lagom.javadsl.persistence.AggregateEventTag
-import com.lightbend.lagom.javadsl.persistence.Offset
+import com.lightbend.lagom.javadsl.persistence.{ Offset => LagomOffset }


I was getting a bit tired of so many Offset types in the code and having to scroll to the imports every time to see what Offset was the code referring to (Lagom's or Akka Persistence Query) so I started using aliases anywhere I would step into.

This change makes the PR bigger than it should but hopefully also makes the PR more readable.

This is really good. Thanks for doing that. It's making everything much easier to read and helps to review this PR.

ignasi35 · 2020-03-06T16:35:43Z

persistence/core/src/main/scala/com/lightbend/lagom/internal/spi/projection/ProjectionSpi.scala

+  @InternalStableApi
+  private[lagom] def completedProcessing(offset: Future[Offset], exCtx: ExecutionContext): Future[Offset] = offset
+
+}


This is the main change in the PR.

private[lagom] def startProcessing(system: ActorSystem, tagName: String, envelope: EventEnvelope):

This is invoked as soon as an event is read from the journal. See AbstractPersistentnEntityRegistry.eventStream

private[lagom] def afterUserFlow(projectionName: String, offset: Offset): Offset = offset

This is invoked, if possible, immediately after completing the user code in the stream. This is sometimes not possible (depends on the projection implementation).

private[lagom] def completedProcessing(offset: Future[Offset], exCtx: ExecutionContext): Future[Offset]

This is invoked when the offset has been persisted.

private[lagom] def failed(actorSystem: ActorSystem, projectionName: String, partitionName: String): ``` This is invoked is a failure traverses the stream on the porjection.

ignasi35 · 2020-03-06T16:36:00Z

.../javadsl/src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/OffsetAdapter.scala

  def offsetToDslOffset(offset: Offset): LagomJavaDslOffset = offset match {
    case TimeBasedUUID(uuid) => LagomJavaDslOffset.timeBasedUUID(uuid)
    case Sequence(value)     => LagomJavaDslOffset.sequence(value)
    case NoOffset            => LagomJavaDslOffset.NONE
-    case _                   => throw new IllegalArgumentException("Unsuppoerted offset type " + offset.getClass.getName)
+    case _                   => throw new IllegalArgumentException("Unsupported offset type " + offset.getClass.getName)


ignasi35 · 2020-03-06T16:36:58Z

.../javadsl/src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/ReadSideActor.scala

+import com.lightbend.lagom.javadsl.persistence.AggregateEvent
+import com.lightbend.lagom.javadsl.persistence.AggregateEventTag
+import com.lightbend.lagom.javadsl.persistence.ReadSideProcessor
+import com.lightbend.lagom.javadsl.persistence.{ Offset => LagomOffset }


This ReadSideActor without the Offset aliases was impossible to reason about. When I added the userFlowWrapper it was too confusing.

ignasi35 · 2020-03-06T16:38:04Z

.../javadsl/src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/ReadSideActor.scala

@@ -53,11 +66,11 @@ private[lagom] object ReadSideActor {
 * Read side actor
 */
 private[lagom] class ReadSideActor[Event <: AggregateEvent[Event]](
-    tagName: String,
+    workerCoordinates: WorkerCoordinates,


This was a pending improvement: having both the projection name and the tag name inside the actor.
The need to report the projection name in the metrics finally twisted my arm into implementing it.

ignasi35 · 2020-03-06T16:39:47Z

.../javadsl/src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/ReadSideActor.scala


          scaladsl.Source
-            .fromFuture(futureOffset)
+            .future(futureOffset)


This was already implemented in #2614 but that PR was wrongfully targeting master. When forward merging 1.6.x into master this will CONFLICT.

I guess not since it's exactly the same change.

ignasi35 · 2020-03-06T16:42:23Z

.../javadsl/src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/ReadSideActor.scala

+
+      unzip.out0 ~> wrappedFlow ~> zip.in0
+      unzip.out1 ~> zip.in1
+      zip.out ~> metricsReporter.in


This is the main change in ReadSide processors. The user-provided flow (which is user code inside a ReadSideHandler so both user statements and OffsetStore statments) is now wrapped and a side-channel for the offset to propagate alongside the userflow exists. The side-channel zips back and produces an Offset for the metrics.

ignasi35 · 2020-03-06T16:43:58Z

project/Dependencies.scala

@@ -28,7 +28,7 @@ object Dependencies {
    val Akka: String = sys.props.getOrElse("lagom.build.akka.version", "2.6.3") // sync with docs/build.sbt
    val AkkaHttp     = "10.1.11"

-    val AkkaPersistenceCassandra = "0.102"
+    val AkkaPersistenceCassandra = "0.103-SNAPSHOT"


Already fixed in #2758

ignasi35 · 2020-03-06T16:44:34Z

...ection/core/src/test/scala/com/lightbend/lagom/internal/projection/ProjectionStateSpec.scala

@@ -2,7 +2,7 @@
 * Copyright (C) Lightbend Inc. <https://www.lightbend.com>
 */

-package com.lightbend.lagom.internal.cluster.projections
+package com.lightbend.lagom.internal.projection


Moved class to the appropriate package. Standardized to using singular.

ignasi35 · 2020-03-06T16:47:10Z

...fka/server/src/main/scala/com/lightbend/lagom/internal/broker/kafka/TopicProducerActor.scala

+                    (tuple._1, o)
+                  }
+                  .via(eventPublisherFlow)
+                  .map(o => ProjectionSpi.completedProcessing(o, context.dispatcher))


Because of how TopicProducer works, the changes in TopicProducerActor differ from the changes in ReadSideActor implementations above.

Here, we don't have a side channel to propagate the Offset because the offset propagates already. But also because in a Topic Flow users may decide to filter events in which case using a side-channel becomes useless.

raboof · 2020-03-30T15:03:58Z

persistence/core/src/main/scala/com/lightbend/lagom/internal/spi/projection/ProjectionSpi.scala

+  private[lagom] def afterUserFlow(projectionName: String, offset: Offset): Offset = offset
+
+  @InternalStableApi
+  private[lagom] def completedProcessing(offset: Future[Offset], exCtx: ExecutionContext): Future[Offset] = offset


This seems slightly strange: is there value in capturing both the moment completedProcessing is called and the moment offset resolves? Or is there something around context propagation that means context must be captured before offset resolves?

I completely missed that. Thanks @raboof

octonato

Thanks @ignasi35. Great we are getting this.

I left a few minor improvements request. I'm approving it already, but added the block-merge label, maybe you are expecting another ✅ from @raboof.

Feel free to remove the label.

octonato · 2020-03-31T09:51:35Z

.../javadsl/src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/ReadSideActor.scala


          scaladsl.Source
-            .fromFuture(futureOffset)
+            .future(futureOffset)


I guess not since it's exactly the same change.

.../javadsl/src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/ReadSideActor.scala

...fka/server/src/main/scala/com/lightbend/lagom/internal/broker/kafka/TopicProducerActor.scala

octonato · 2020-03-31T10:16:17Z

...la/com/lightbend/lagom/internal/javadsl/persistence/cassandra/CassandraReadSideHandler.scala

@@ -21,7 +21,7 @@ import com.lightbend.lagom.javadsl.persistence.ReadSideProcessor.ReadSideHandler
 import com.lightbend.lagom.javadsl.persistence.cassandra.CassandraSession
 import com.lightbend.lagom.javadsl.persistence.AggregateEvent
 import com.lightbend.lagom.javadsl.persistence.AggregateEventTag
-import com.lightbend.lagom.javadsl.persistence.Offset
+import com.lightbend.lagom.javadsl.persistence.{ Offset => LagomOffset }


This is really good. Thanks for doing that. It's making everything much easier to read and helps to review this PR.

ignasi35 force-pushed the projections-spi branch from 00b7388 to 9dc2be2 Compare March 6, 2020 16:28

ignasi35 commented Mar 6, 2020

View reviewed changes

ignasi35 added 4 commits March 19, 2020 21:23

Introduce Projections Spi

602ec7b

Use singular for projection. Use APC Snapshot. Spi APi review

fb05f2c

Adds side-channel for ReadSideActor (scaladsl)

2780e38

Adds side channel for ReadSideActor (javadsl). Other code cleanups

8ed60c4

ignasi35 force-pushed the projections-spi branch from 9dc2be2 to 8ed60c4 Compare March 19, 2020 20:23

Commit track event processing failures

01ab9f9

ignasi35 mentioned this pull request Mar 27, 2020

Small improvements to tests Lagom projections telemetry lightbend/telemetry-samples#3

Closed

ignasi35 marked this pull request as ready for review March 27, 2020 18:42

ignasi35 requested a review from octonato March 27, 2020 18:42

ignasi35 added 2 commits March 27, 2020 19:44

Fix compilation error.

c64a7ee

MiMa filters ignore all *.interna.*

ea71401

raboof reviewed Mar 30, 2020

View reviewed changes

ignasi35 added 2 commits March 30, 2020 17:40

Propagate teh exception to telemetry in case of failures

d7b18a9

refactor to remove PorjectionContext from telemetry

2eb082d

octonato added the status:block-merge label Mar 31, 2020

octonato approved these changes Mar 31, 2020

View reviewed changes

ignasi35 changed the title ~~[WIP] Introduce Projections spi~~ Introduce Projections spi Mar 31, 2020

ignasi35 added 2 commits March 31, 2020 16:05

PR comments

ab927a9

Simplify SPI (thanks @raboof)

45c906a

octonato removed the status:block-merge label Apr 1, 2020

octonato merged commit 4aa1247 into lagom:1.6.x Apr 1, 2020

octonato added this to the 1.6.2 milestone Apr 23, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce Projections spi #2727

Introduce Projections spi #2727

ignasi35 commented Feb 21, 2020

ignasi35 Mar 6, 2020

octonato Mar 31, 2020

ignasi35 Mar 6, 2020

ignasi35 Mar 27, 2020

ignasi35 Mar 6, 2020

ignasi35 Mar 6, 2020

ignasi35 Mar 6, 2020

ignasi35 Mar 6, 2020

octonato Mar 31, 2020

ignasi35 Mar 6, 2020

ignasi35 Mar 6, 2020

ignasi35 Mar 6, 2020

ignasi35 Mar 6, 2020

raboof Mar 30, 2020

ignasi35 Apr 1, 2020

octonato left a comment

octonato Mar 31, 2020

octonato Mar 31, 2020

Introduce Projections spi #2727

Introduce Projections spi #2727

Conversation

ignasi35 commented Feb 21, 2020

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

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

octonato left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment