Saves offset for unhandled events #915

octonato · 2017-08-04T12:32:12Z

Fixes #904

octonato · 2017-08-04T12:56:46Z

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

+            if (log.isDebugEnabled) log.debug("Unhandled event [{}]", pair.first.getClass.getName)
+            executeStatements(offsetStatement(pair.second))
+        }
+


Instead of this pattern matching, I would have prefer to work with a NoOps handler emitting an empty list of BoundStatement, but I couldn't manage to build one. I'm get some compilation error (won't give the details here).

A NoOps handler is a much cleaner solution because it will increase local reasoning. The following would be much simple implementation.

val handler = handlers.getOrElse( pair.first.getClass.asInstanceOf[Class[Event]]), NoOpsHandler ) invoke(handler, pair.first, pair.second).toScala.flatMap(executeStatements)

The invoke method would always add the offset statement and we avoid the branching and excessive logic.

Found a way to encode an emptyHandler. Will push it soon.

(resolves lagom#904)

octonato · 2017-08-04T14:05:08Z

.../src/main/scala/com/lightbend/lagom/internal/javadsl/persistence/jdbc/JdbcReadSideImpl.scala

+              // whatever it happens we save the offset
+              offsetDao.updateOffsetQuery(OffsetAdapter.dslOffsetToOffset(pair.second))
+            }
+            .map(_ => Done)


Live is so much easier with a good IO monad around.

octonato · 2017-08-04T14:24:19Z

It's really a shame that there is so much code duplication here. All classes are doing more or less the same, except the JPA which is a beast apart.

The only difference between them is the usage of collections and Futures. The java versions are in Scala, but use CompletionStage, java Collections and Pair. The Scala versions have Future, scala Collection and EventStreamElement.

Obviously there are some crucial differences between the JDBC (Slick) and Cassandra, but still there are a lot of things in common.

octonato · 2017-08-07T14:58:23Z

...ce/javadsl/src/test/scala/com/lightbend/lagom/javadsl/persistence/AbstractReadSideSpec.scala

+      count shouldBe expected
+    }
+  }
+


Previous versions was using some methods coming from the akka-testkit. It was a combination of within, awaitAssert and Await.result.

I find the ScalaTest DSL for this kind of async test much cleaner. It also reuses the timeout (PatienceConfig) everywhere in the test. We don't need to repeat over and over again.

octonato · 2017-08-07T14:59:23Z

...ce/javadsl/src/test/scala/com/lightbend/lagom/javadsl/persistence/AbstractReadSideSpec.scala

-        count should ===(expected)
-      }
-    }
+  private val tag = TestEntity.Evt.AGGREGATE_EVENT_SHARDS.forEntityId("1")


moved tag to instance val because it's also used on the fetchLastOffset method

octonato · 2017-08-07T15:00:21Z

...ce/javadsl/src/test/scala/com/lightbend/lagom/javadsl/persistence/AbstractReadSideSpec.scala

-    }
+  private val tag = TestEntity.Evt.AGGREGATE_EVENT_SHARDS.forEntityId("1")
+
+  private def createTestEntityRef() = {


CreateEntityRef gets it's own create method.

…ther

TimMoore · 2017-08-09T00:23:57Z

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

+        val handler =
+          handlers.getOrElse(
+            // lookup handler
+            pair.first.getClass.asInstanceOf[Class[Event]],


I think the cast is not needed (IDK why it was there in the original code)

It also might be nice to destructure the pair at the top of the mapAsync block... I find with pair.first, pair.second that I have to keep referring back to understand what each one is.

yep, extracting is a great idea

TimMoore · 2017-08-09T00:33:39Z

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

+          val batch = new BatchStatement
+          val iter = statements.iterator()
+          while (iter.hasNext)
+            batch.add(iter.next)


BatchStatement has an addAll(Iterable) method... not sure why this wasn't using it in the first place.

didn't now it. Will change for sure

TimMoore · 2017-08-09T00:48:00Z

...ce/javadsl/src/test/scala/com/lightbend/lagom/javadsl/persistence/AbstractReadSideSpec.scala

+      .buildHandler()
+      .prepare(tag)
+      .toScala
+      .map(_.asInstanceOf[O]) // no ClassTag, no mapTo :-(


Why even parameterize the offset type? AFAICT, nothing actually cares about the type.

I guess it comes from my initial wrong assumption that is was always a Long. I had a mapTo[Long] that I had to removed because cassandra was returning a TimeBasedUUID.

But you are right, we just keep it as Any. That's all private and internal usage.

Or mapTo[Offset], just because I feel bad to return Any. ;-)

TimMoore · 2017-08-09T00:54:31Z

.../scaladsl/src/test/scala/com/lightbend/lagom/scaladsl/persistence/AbstractReadSideSpec.scala


    readSide
  }

+  private def assertSelectCount(id: String, expected: Long): Unit = {
+    within(20.seconds) {
+      awaitAssert {


Did you mean to switch this to use eventually like in the Java version?

yes, sure. I missed this one

TimMoore · 2017-08-09T00:55:51Z

...ce/javadsl/src/test/scala/com/lightbend/lagom/javadsl/persistence/AbstractReadSideSpec.scala

+      // persisted offset must have change tough
+      watch(readSide)
+      system.stop(readSide)
+      expectTerminated(readSide)


Should all of this be in a finally? If the eventually fails, I guess it won't stop the read side here.

Good point. I will mixin the BeforeAndAfter trait and add it to an after block instead. I'm afraid that a try/finally will add unnecessary noisy to the test method itself.

We are already pushing infrastructure code to helper methods (createReadSideProcessor and createTestEntityRef) .

ignasi35 · 2017-08-10T10:04:34Z

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

+            }
+          )
+
+        invoke(handler, event, offset).toScala.flatMap(executeStatements)


nice extract!

ignasi35 · 2017-08-10T10:07:00Z

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

+        TreePVector
+          .from(statements)
+          .plus(offsetDao.bindSaveOffset(akkaOffset))
+          .asInstanceOf[JList[BoundStatement]]


formatting win.

ignasi35 · 2017-08-10T10:20:56Z

...ce/javadsl/src/test/scala/com/lightbend/lagom/javadsl/persistence/AbstractReadSideSpec.scala


 import scala.compat.java8.FutureConverters._
 import scala.concurrent.Await
 import scala.concurrent.duration._

-trait AbstractReadSideSpec extends ImplicitSender { spec: ActorSystemSpec =>
+trait AbstractReadSideSpec extends ImplicitSender with ScalaFutures with Eventually with BeforeAndAfter { spec: ActorSystemSpec =>


octonato · 2017-08-10T10:56:30Z

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

+
+    def executeStatements(statements: JList[BoundStatement]): Future[Done] =
+      statements.size match {
+        case 0 => Future.successful(Done.getInstance())


Actually statements.size is never 0 because when we fallback to an empty handler, we always add at least one statement (save offset) to the list.

octonato · 2017-08-10T10:59:33Z

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

+    def executeStatements(statements: JList[BoundStatement]): Future[Done] =
+      statements.size match {
+        case 0 => Future.successful(Done.getInstance())
+        case 1 => session.executeWrite(statements.get(0)).toScala


We could also argue that since statements are never empty, we could simply not check the size and always use BatchStatement. The code would be cleaner in that case with a small penalty of running a batch of one single statement.

def executeStatements(statements: JList[BoundStatement]): Future[Done] = { val batch = new BatchStatement batch.addAll(statements) session.executeWriteBatch(batch).toScala }

LGTM, we can add that reasoning as a comment and simplify the code.

TimMoore · 2017-08-11T10:57:53Z

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

+
+    def executeStatements(statements: JList[BoundStatement]): Future[Done] = {
+      val batch = new BatchStatement
+      // statements is never empty, there is at least the store offset statement


I think it can be empty for LegacyCassandraReadSideHandler.

Is it a problem if you do execute an empty batch or will it just handle as a no-op internally?

good point. I will check it

I couldn't out if this could be a problem, so added a if/else check.

This only affect the javadsl because the scala one does not have a legacy version

ignasi35 · 2017-08-11T13:42:05Z

While backporting I had to remove the work done on the SlickReadSide which is a feature not available in 1.3.x.

* saves offset for unhandled events (resolves #904) * adds logging back for unhandled events * saves offset for unhandled events for JDBC, JPA and Slick * slightly improved JPA version * just formatting * Extended tests for storing offsets on unhandled events * bring similater code together - count and offset check next to each other * typo * formatting * uses `batch.addAll` api + extracting `Pair` * removes offset type parameter * controller shutdowns of read side after each test * Using duration's .seconds and .millis instead of ScalaTest Span * using Batch API in all cases * checking for empty statements

ignasi35 · 2017-08-11T13:43:32Z

backport 1.3.x 8417ce6

* saves offset for unhandled events (resolves #904) * adds logging back for unhandled events * saves offset for unhandled events for JDBC, JPA and Slick * slightly improved JPA version * just formatting * Extended tests for storing offsets on unhandled events * bring similater code together - count and offset check next to each other * typo * formatting * uses `batch.addAll` api + extracting `Pair` * removes offset type parameter * controller shutdowns of read side after each test * Using duration's .seconds and .millis instead of ScalaTest Span * using Batch API in all cases * checking for empty statements

octonato force-pushed the read-side-offset-persistence branch from 9f51ac1 to 9eb467c Compare August 4, 2017 12:33

octonato requested a review from TimMoore August 4, 2017 12:34

octonato added status:in-progress status:needs-backport labels Aug 4, 2017

octonato commented Aug 4, 2017

View reviewed changes

saves offset for unhandled events

b96a759

(resolves lagom#904)

octonato force-pushed the read-side-offset-persistence branch from 9eb467c to b96a759 Compare August 4, 2017 13:27

Renato Cavalcanti added 2 commits August 4, 2017 15:37

adds logging back for unhandled events

9dc4e61

saves offset for unhandled events for JDBC, JPA and Slick

9eed3c9

octonato commented Aug 4, 2017

View reviewed changes

Renato Cavalcanti added 2 commits August 4, 2017 16:11

slightly improved JPA version

6295d6e

just formatting

2c32788

Extended tests for storing offsets on unhandled events

ec34e50

octonato changed the title ~~[WIP] Saves offset for unhandled events~~ Saves offset for unhandled events Aug 7, 2017

octonato added status:to-review and removed status:in-progress labels Aug 7, 2017

octonato commented Aug 7, 2017

View reviewed changes

Renato Cavalcanti added 3 commits August 7, 2017 17:04

bring similater code together - count and offset check next to each o…

97cfaec

…ther

typo

73f3d44

formatting

8398c9e

TimMoore reviewed Aug 9, 2017

View reviewed changes

octonato requested a review from ignasi35 August 9, 2017 08:44

Renato Cavalcanti added 3 commits August 9, 2017 11:39

uses batch.addAll api + extracting Pair

45f7746

removes offset type parameter

b4287de

controller shutdowns of read side after each test

95d4fa4

octonato added this to the 1.3.7 milestone Aug 10, 2017

ignasi35 reviewed Aug 10, 2017

View reviewed changes

ignasi35 approved these changes Aug 10, 2017

View reviewed changes

Using duration's .seconds and .millis instead of ScalaTest Span

ae2745a

octonato commented Aug 10, 2017

View reviewed changes

using Batch API in all cases

891ea81

ignasi35 approved these changes Aug 10, 2017

View reviewed changes

TimMoore reviewed Aug 11, 2017

View reviewed changes

checking for empty statements

0c72003

ignasi35 merged commit 03b4ec7 into lagom:master Aug 11, 2017

marcospereira removed the status:to-review label Aug 11, 2017

ignasi35 added type:improvement and removed status:needs-backport labels Aug 11, 2017

octonato deleted the read-side-offset-persistence branch December 22, 2017 08:41

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Saves offset for unhandled events #915

Saves offset for unhandled events #915

octonato commented Aug 4, 2017 •

edited

octonato Aug 4, 2017

octonato Aug 4, 2017

octonato Aug 4, 2017 •

edited

octonato commented Aug 4, 2017

octonato Aug 7, 2017

octonato Aug 7, 2017

octonato Aug 7, 2017

TimMoore Aug 9, 2017

octonato Aug 9, 2017

TimMoore Aug 9, 2017

octonato Aug 9, 2017

TimMoore Aug 9, 2017

octonato Aug 9, 2017

octonato Aug 9, 2017

TimMoore Aug 9, 2017

octonato Aug 9, 2017

TimMoore Aug 9, 2017

octonato Aug 9, 2017

ignasi35 Aug 10, 2017

ignasi35 Aug 10, 2017

ignasi35 Aug 10, 2017

octonato Aug 10, 2017

octonato Aug 10, 2017

ignasi35 Aug 10, 2017

TimMoore Aug 11, 2017

TimMoore Aug 11, 2017

octonato Aug 11, 2017

octonato Aug 11, 2017

ignasi35 commented Aug 11, 2017

ignasi35 commented Aug 11, 2017

Saves offset for unhandled events #915

Saves offset for unhandled events #915

Conversation

octonato commented Aug 4, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

octonato Aug 4, 2017 • edited

Choose a reason for hiding this comment

octonato commented Aug 4, 2017

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

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

ignasi35 commented Aug 11, 2017

ignasi35 commented Aug 11, 2017

octonato commented Aug 4, 2017 •

edited

octonato Aug 4, 2017 •

edited