Recovery strategies when handler fails to process, #17 #95

patriknw · 2020-05-07T09:20:29Z

moved recovery strategies and implementation to core
used in CassandraProjection (and SlickProjection)
new unit test of the HandlerRecoveryImpl
also found and fixed a few bugs
- problem with at-least-once after 1 in Slick
- problem with at-most-once in Cassandra, async boundary may cause more than one
  offset to be saved before processing

References #17

Draft because on top of other PR, but ready for review.

patriknw · 2020-05-07T09:22:10Z

...on-cassandra/src/main/scala/akka/projection/cassandra/internal/CassandraProjectionImpl.scala

+            case (offset, envelope) =>
+              offsetStore
+                .saveOffset(projectionId, offset)
+                .flatMap(_ => applyUserRecovery(envelope, () => handler.process(envelope)))


It was wrong because the async boundary may cause more than one offset to be saved before processing

patriknw · 2020-05-07T09:22:47Z

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickProjectionImpl.scala


    // TODO: add a LogSource for projection when we have a name and key
    val logger = Logging(systemProvider.classicSystem, this.getClass)

-    val futDone = Future.successful(Done)
-
-    def applyUserRecovery(envelope: Envelope)(futureCallback: () => Future[Done]) = {


this was moved to core (and improved)

patriknw · 2020-05-07T09:25:20Z

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickProjectionImpl.scala

-      }
-    }
-
-    def aroundUserHandler(env: Envelope)(handlerFunc: Envelope => slick.dbio.DBIO[Done]) = {


I removed this. I think the purpose was to deal with thrown exceptions, but that is now handled in same way as if the handler returned Future.failed. There were no tests for this and I don't think it was completely correct anyway.

patriknw · 2020-05-07T09:27:15Z

akka-projection-slick/src/main/scala/akka/projection/slick/internal/SlickProjectionImpl.scala

          .transactionally

      applyUserRecovery(env) { () =>
        databaseConfig.db.run(txDBIO).map(_ => Done)
      }
    }

-    def processEnvelopeAndStoreOffsetInSeparateTransactions(env: Envelope): Future[Done] = {


removed this, and using future composition instead, see the AtLeastOnce(1, _) case below. Strange if offsetStore fails, then we would retry and run the handler transaction again.

octonato

LGTM

A few change request here and there though

octonato · 2020-05-07T12:22:44Z

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

+  def applyUserRecovery[Offset, Envelope](
+      handler: HandlerRecovery[Envelope],
+      envelope: Envelope,
+      sourceProvider: SourceProvider[Offset, Envelope],


here we can pass just the extracted offset

octonato · 2020-05-07T12:23:57Z

akka-projection-core/src/main/scala/akka/projection/javadsl/Handler.scala

@@ -17,6 +29,6 @@ import akka.Done
 * guarantees between the invocations are handled automatically, i.e. no volatile or
 * other concurrency primitives are needed for managing the state.
 */
-trait Handler[Envelope] {
+abstract class Handler[Envelope] extends HandlerRecovery[Envelope] {


@ApiMayChange like in Scala?

yes, I'll also add a ticket so that we take a look at all of it before "final"

octonato · 2020-05-07T12:26:46Z

akka-projection-core/src/main/scala/akka/projection/HandlerRecovery.scala

+    retryAndSkip(retries, delay.asScala)
+
+  /**
+   * INTERNAL API: placed here instead of the `internal` package because of sealed trait


octonato · 2020-05-07T13:45:46Z

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

+            retied
+
+          case RetryAndSkip(retries, delay) =>
+            logger.error(


warning since we are skipping

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

octonato · 2020-05-07T13:54:21Z

akka-projection-core/src/test/scala/akka/projection/internal/HandlerRecoveryImplSpec.scala

+    "fail" in {
+      val handler = new FailHandler(HandlerRecoveryStrategy.fail, failOnOffset = 3)
+      val result =
+        HandlerRecoveryImpl.applyUserRecovery(handler, env3, sourceProvider, logger, () => handler.process(env3))


if we pass only the offset, we don't even need to hack a SourceProvider here.

👍

The projection impl will have access to the Envelope and SourceProvider already so it can extract the offset then.

octonato · 2020-05-07T13:58:28Z

akka-projection-slick/src/test/scala/akka/projection/slick/SlickProjectionSpec.scala

+
+  class ConcatHandlerFail4(recoveryStrategy: HandlerRecoveryStrategy = HandlerRecoveryStrategy.fail)
+      extends SlickHandler[Envelope] {
+    private val _attempts = new AtomicInteger()


👍, I didn't think about making this atomic on first impl. Good catch.

reason is that it is accessed from the outside of the handler, from the test

patriknw · 2020-05-07T14:26:20Z

changed so that retries are not used for atMostOnce because it would be more than once

seglo

Really nice 👍

seglo · 2020-05-07T15:06:42Z

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

+
+            // retries - 1 because retry() is based on attempts
+            // first attempt is performed immediately and therefore we must first delay
+            val retied = after(delay, scheduler)(retry(tryFutureCallback, retries - 1, delay))


seglo · 2020-05-07T15:08:41Z

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

+            futDone
+
+          case RetryAndFail(retries, delay) =>
+            logger.error(


Should this be a warning since we're going to retry?

yeah, I think so

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

seglo · 2020-05-07T15:12:48Z

akka-projection-core/src/test/scala/akka/projection/internal/HandlerRecoveryImplSpec.scala

+    "fail" in {
+      val handler = new FailHandler(HandlerRecoveryStrategy.fail, failOnOffset = 3)
+      val result =
+        HandlerRecoveryImpl.applyUserRecovery(handler, env3, sourceProvider, logger, () => handler.process(env3))


👍

The projection impl will have access to the Envelope and SourceProvider already so it can extract the offset then.

* moved recovery strategies and implementation to core * used in CassandraProjection (and SlickProjection) * new unit test of the HandlerRecoveryImpl * also found and fixed a few bugs * problem with at-least-once after 1 in Slick * problem with at-most-once in Cassandra, async boundary may cause more than one offset to be saved before processing

patriknw · 2020-05-07T18:25:29Z

incorporated feedback

patriknw · 2020-05-08T06:36:51Z

some test failure, I'll look into it

* simple search and replace refactoring so not much too review

patriknw commented May 7, 2020

View reviewed changes

octonato mentioned this pull request May 7, 2020

Handler trait in Cassandra Projection, #43 #93

Merged

octonato approved these changes May 7, 2020

View reviewed changes

seglo approved these changes May 7, 2020

View reviewed changes

patriknw force-pushed the wip-43-eventhandler-trait-patriknw branch from 4707a78 to d74431d Compare May 7, 2020 17:37

patriknw added 3 commits May 7, 2020 20:24

don't use retries for atMostOnce

363f1b0

review feedback

3b38393

patriknw force-pushed the wip-17-fail-strat-patriknw branch from a6c58b1 to 3b38393 Compare May 7, 2020 18:24

patriknw changed the base branch from wip-43-eventhandler-trait-patriknw to master May 7, 2020 18:25

patriknw marked this pull request as ready for review May 7, 2020 18:25

patriknw closed this May 7, 2020

patriknw reopened this May 7, 2020

patriknw mentioned this pull request May 8, 2020

Kafka SourceProvider implementation and Slick integration example #80

Merged

fix bug in RetryAndSkip

251b5cc

patriknw merged commit 42d4bf0 into master May 8, 2020

patriknw deleted the wip-17-fail-strat-patriknw branch May 8, 2020 09:37

patriknw mentioned this pull request May 11, 2020

Revisit database concurrency in KafkaToSlickIntegration example #107

Closed

johanandren pushed a commit that referenced this pull request Mar 23, 2023

Remove hint from entity_type_hint, #95

dcba803

* simple search and replace refactoring so not much too review

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Recovery strategies when handler fails to process, #17 #95

Recovery strategies when handler fails to process, #17 #95

patriknw commented May 7, 2020

patriknw May 7, 2020

patriknw May 7, 2020

patriknw May 7, 2020

patriknw May 7, 2020

octonato left a comment

octonato May 7, 2020

patriknw May 7, 2020

octonato May 7, 2020

patriknw May 7, 2020

octonato May 7, 2020

octonato May 7, 2020

octonato May 7, 2020

seglo May 7, 2020

octonato May 7, 2020

patriknw May 7, 2020

patriknw commented May 7, 2020

seglo left a comment

seglo May 7, 2020

seglo May 7, 2020

octonato May 7, 2020

seglo May 7, 2020

patriknw commented May 7, 2020

patriknw commented May 8, 2020

Recovery strategies when handler fails to process, #17 #95

Recovery strategies when handler fails to process, #17 #95

Conversation

patriknw commented May 7, 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

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

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

patriknw commented May 7, 2020

seglo 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

patriknw commented May 7, 2020

patriknw commented May 8, 2020