Down-streaming with Event Sourcing application.

How to apply side effects in Event Sourcing application with desirable/expected consistency level?
It's common and logical to put side effects as a Sink/End of Persistent stream. Usually side effects are dispatched in event-persisted callback handlers. These handlers can also spawn new one sub-streams to react on applied events.

There are different tactics to integrate down-streams to the business flow and this is serious architectural question that makes influence on all the application's design layers.

To review all the possibilities I'm going to use an example of typical event-sourcing application. Let's assume it implements some business utility.  After each event is dispatched we are sending the notification to Business Intelligence services for analyzing and reporting. Imagine the day when you find out - that Business Intelligence operations are fast enough and can produce real-time feedback which can increase the profitability of our business up to X%. We are using BI as an example and placeholder - but in common it can be any downstream service reacting to our business events. Down-stream service are the ones that consume the upstream service.
Just to rephrase it and agree on terminology - I will use BI in all the examples - but it could be any down-stream service.

Given the terminology 'upstream' and 'downstream' it may help to make an analogy with a river. If you drop a message (data) in the river it flows from upstream (initiator) to downstream (receiver).

Here is the typical event-sourcing flow from request to response including other side effects:

Figure 1. Typical event souring flow

After stream emits a command - it should:

1. Validate input against the current state and business rules then generate mutation events.
2. Persist an event(s)/state.
3. Respond to event producer if needed.
4. Execute other important side-effects
5. Send events/messages/state to BI (down-stream)

Here is pseudo-code that implements the behavior using akka-persistence:

It's quite logical to apply all side-effects after the mutation has been already persisted (as we operates up-stream and down-stream patterns). In step 3 we send response to the client. Our actor is abstracted from delivering this message, in particular case we rely to "at most once" akka's guarantee. If client expects the different one then we should build business protocol to provide extra guarantees - for example handle timeouts etc.

Here is sequence diagram that helps us to find the flow's operations with different consistency levels:

Figure 2. Consistency per stream flow

The legend for Figure 2 is:

• A to B is under your API interface consistency - for example it could be Web-sockets protocol with some custom retry policy. We are going to ignore this step flow in examples, ideate it has tolerable consistency and our Architecture design responsible only for the flow from B to E. 
• C during persistency execution we rely on our persistence journal's consistency - for example Cassandra's.
• C to D - After journal persisted event - the flow returns to app. "at least once".
• E - is downstream - it can start the new sub-stream with A to D steps or rely on some other delivery tools, for example Kafka.

As we can see the weakest consistency is between C and D and it determines consistency to BI down-stream. Actually our example implements the first solution with the lowest consistency level but it has own pros as well.

Figure 3. BI side-effects with at most once guarantee

Solution #1: At-most-once, i.e. no guaranteed delivery 

If we send events to BI after the persistence step:

• The response time from client perspective doesn't depend on down-streaming performance
• BI delivery failures don't influence on our business flow
• BI can miss out events (at most once delivery guarantee)
• Low latency delivery to BI downstream

We can miss out downstream events - for example if we use Kafka producer and rely on Kafka's persistence - when producer.flush operation is failed we might lose the events.

Solution #2: Possible redundant events

Figure 4. Allow redundant events 

If the fact of receiving the events has a priority over dispatching them - it's possible to change the guarantee to "at least once" allowing the duplicates (in case of client retries) and inconsistency between (not all the events are persisted - but all are reported).

This solution provides:

• Better down-stream but worse response latency (persist waits the flush for downstream completed).
• At least one guarantee for downstream - but possible inconsistency to state.
• Can be combined with solution #1 when different events have different guarantees.

Solution #3: Connector to persistence layer

Figure 5. Connector to persistence layer.

Sometimes down-streaming the data is becoming critical for the business flow and more reliable guarantees are wanted. It's possible to use the tools that allow to subscribe for persistence and convert them into stream.

There are plenty of market ready solution.  For example if Cassandra is used as persistence for your journal and BI is a Kafka's consumer - then you can look for different Kafka to Cassandra connectors.  In case of Akka is standing for event-sourcing in your design  - look into Persistence Query

The difference to solution #1 is:

• Better delivery guarantee to BI (could be tuned to at least once or exact once)
• Decline in latency

Solution is primely except the cases when latency of BI events is critical. 

You can try to implement custom solution - for example for Cassandra it's possible to read commit log and send the entries into Kafka. It will allow you to tune the latency between Cassandra and Kafka on the lowest possible level - but it will be always a tradeoff - the quicker you make Cassandra flush the data to journal - the slower will be your persistence - but faster stream to BI.

Solution #4 Integrate into persist step

When down-streaming the events is critical part of your flow and not been able to proceed with it means unrecoverable failure for application - then it's logical to amalgamate persist action with down-streaming.

Figure 6. Integrate into persist step

It means we either persist and send events downstream or fail. In case of akka-persistence we can use the journal that persists the events and sends them to Kafka. This combination can be met with DuplicatingJournal and StreamToKafkaJournal.

This solution is:

• Failing the flow if either persist or send event failures (of course each of them can handle some retry policies etc.)
• Has a delivery guarantee equaled to chosen downstreaming tool (for Kafka can have "at least once" and "exact once"
• Response and other side effects might have worse latency than in solution #1 because of persistTime = max(journalPersistDuration, eventSendFlushDuration)
• Doesn't implement Atomicity guarantees - but explicitly fails in case of inconsistency.

Solution #5 Custom protocol

Let's try to solve the the case when down-streaming consistency and latency are evenly critical - but we can accept eventual consistency in case of failures and quasi real-time for the ninety-nine out of a hundred.
This means we want to continue business flow execution even if down-stream is failing - but we rely on eventual consistency of down-stream - it should recover and continue from the moment it failed as soon as it has been recovered. 

Figure 7. Custom protocol

This is the most expensive solution as it requires implementation of manual delivery protocol to down-stream. Unfortunately Akka's At-Least-Once Delivery is insufficient for our case.
As an example we will use finite-state machine for BI consumer and coupling it's logic to the business flow.

Lets look in details:

1. Command is emitted.
2. Each command after validation is checked is it important to be an initialization marker for business flow. Examples of initialization commands are: start of transaction, user creates shopping card, online game round started etc. Initialization event should be delivered with the most desirable guarantee - but it should not make huge impact on main flow latency - because they take only small percentage from all the events.
3. If event is initial then it has delivery guarantee at least once and business flow just awaits down-stream flush successfully completed. If the event isn't initial then it sent to down-stream in "fire and forget" mode.
4. Events are persisted.
5. Side-effects are applied and one of them is sending message to BI. Fire and forget mode like in solution #1.
6. Sending the response to client.
7. The most complicated part is BI - it has implements finite state machine that validates the messages been received. 

As an example of that solution we can implement FSM protocol for the service that can adds together integer values and support commands:

• InitializeTransaction
• Operand(Int)
• GetState

Rough algorithms is:

1. Business flow receives "InitializeTransaction" command
2. Because we marked this command as initial - it should be sent to BI with at least once consistency.
3. BI awaits with timeout for the next event - EventPersisted(TransactionInitialized) - in case of timeout it should explicitly fetch the state of transaction from the business flow.
4. After event is produced by command is persisted - business flow should send the event to BI with fire and forget style (no delivery guarantee).
5. BI after each event sets timeouts for expecting the next one or final state - it guarantees the eventual consistency between business service and BI down-stream.
6. BI iterates all the 3 to 5 steps until it gets the final marker - transaction completed.

Solution #6 Kafka streams

Use Kafka streams. Kafka streams became self-sufficient platform for building event sourcing applications and doesn't require hybridizing with other frameworks like akka-persistence.

Summary:

If you don't have to rely on strict delivery guarantees for down-streams use a mix of Solution #1: At-most-once and Solution #2: Possible redundant events. 

If delivery latency isn't critical - but consistency is - use Solution #3: Connector to persistence layer. It should fit to the most of business cases and supported out of the box by the most of event-sourced frameworks. 

Solution #4 Integrate into persist step emulates transaction without rollback possibility and is suitable only for special business cases.

When down-streaming latency is critical Solution #5 Custom protocol or Solution #6 Kafka streams are the ways to go. Implementation and support of custom protocol is the most expensive, additionally it validates your main business flow and theoretically can provide good monitoring feedback. Using the Kafka streams has own limitations and isn't suitable for all the business domains.

ScalaCache - conditional caching

Sometimes it's important to avoid caching some subset from return values. It's easy to implement it using Memoization method memoizeF. When using M[_] like Future - the Failed case won't be cached - and it's possible to convert any value to Failed with special exception marker that contains the value itself and in meantime blocking it from been cached.

Sometimes it's preferable to avoid caching some subset of possible values without deviation to failed case of higher kind wrapper. In case if this condition can be delineated by predicate and you don't want to play with implicit mode: Mode[F] you can mixing small trait to your cache:

This example is based on Caffeine and it isn't caching negative integer values:

javax.ws.rs.container.ContainerRequestFilter detracts @Suspended

If your are still abused to use Java API for RESTful Web Services but building non blocking api with help of:

    @Suspended asyncResponse: AsyncResponse

Keep in mind that javax.ws.rs.container.ContainerRequestFilter doesn't support non blocking nature and in case it's a part of invocation chain - it will block the calling thread and make usage of @Suspended

The best workaround is to migrate into functional style code - and use loan pattern if possible.

Code can look like:

    @GET
    @Produces("text/html")
    def handleRequest(@Suspended res: AsyncResponse): Unit = {
      authenticateBasicAsync(realm = "secure site") {
        userName =>
          res.resume(s"The user is '$userName'")
      }
    }

This code will be easier migrate in future to akka-http or other async http libraries.

Сheat sheet: Team assessment

Here is the summary of my experience on what every team should evade or at least minimize. It's agnostic to methodology or it's absence on the project - just set of common sense advises.

It helps assist the team during on-boarding or check it periodically - for example on retrospective meetings.

Team should avoid:

• Unclear goals
• When the methodology for achieving the goal has not been agreed upon.
• Absence of task priorities.
• Inability to tell justified "No" to extra load.
• A large number of tasks that require high concentration.
• The prospects of the tasks aren't visible (when the ultimate goals of the task are unclean).
• Lack of motivation.
• Noise in the working space.
• Insufficient coordination between the colleagues.
• Insufficient delegation of work.
• Insufficiently organized storage of information (and knowledge sharing).
• Too many of meetings.

Fix known problems before writing new code.

It's hard to predict the price of keeping the build in "red" state - especially in micro-services environment. In chaotic world the risk to get unpredictable devastating effects because of inability to deploy/test the latest source code changes are growing in time exponentially.

The same is for the bugs related either to internal processes or product itself. If your project isn't hardly using the Business Intelligence, operates with hundreds live metrics to measure bug impact - assume that defects has a priorities over the features from you backlog. In the worst case scenario teams are introducing the ceremonies to address workarounds for known bugs, because those bugs aren't the priorities.

Bugs aren't always reflected in your kpi(s) or metrics even if your think that BI model is close to real feedback - Customer Satisfaction is something that has a huge lag in measurements.

I found that team using their own product gives higher priorities to quality over adding new features.

For the business perspective in long term "eliminate known defects before working on new features" rule has more benefits.  It allows to keep the project under control with lower price. As long as team accepts the know problems as a risks - they usually forget that incremental operation on risks probabilities is following the rule of multiply the individual probabilities and keep the project to unmanageable way in case "shadow" abruptness.

From technical perspective it's easier to follow those rule as long as continues integration rule requires:

 Fix Broken Builds Immediately.

Team just needs to add the extra rule:

 Fix know problems before writing new code

Speedup tests execution in ScalaTest

Sometimes the duration of automated tests written with ScalaTests library can be reduced in times easly.

Here are the few steps I found to make it fast and easier.


  1. Maven: use the latest version of maven itself 3.5.x and scala-maven-plugin 3.4.x Be careful with other plugins aren't supporting parallel execution, if you see this message in build log, carefully review the plugins list

[WARNING] *****************************************************************
[WARNING] * Your build is requesting parallel execution, but project *
[WARNING] * contains the following plugin(s) that have goals not marked *
[WARNING] * as @threadSafe to support parallel building. *

  2. Switch on parallel mode and disable JVM forking:

  Maven and sbt:

ScalaTest's normal approach for running suites of tests in parallel is to run different suites in parallel, but the tests of any one suite sequentially. This approach should provide sufficient distribution of the work load in most cases, but some suites may encapsulate multiple long-running tests. Such suites may dominate the execution time of the run. If so, mixing in this trait into just those suites will allow their long-running tests to run in parallel with each other, thereby helping to reduce the total time required to run an entire run.

  3. Analyse the shared state. Switching the tests into parallel mode sometimes introduces race conditions. If test methods are sharing mutable state in class fields. For example:
The best solution is usage of shared fixtures or test scopes. If you don't want to refactor your test in BigBang approach it's easier to extend the Suits/Specs with org.scalatest.OneInstancePerTest.

Trait that facilitates a style of testing in which each test is run in its own instance of the suite class to isolate each test from the side effects of the other tests in the suite. If you mix this trait into a Suite, you can initialize shared reassignable fixture variables as well as shared mutable fixture objects in the constructor of the class. Because each test will run in its own instance of the class, each test will get a fresh copy of the instance variables. This is the approach to test isolation taken, for example, by the JUnit framework.

  4. Reduce the number of heavy resources via re-sharing them. Sometimes there could be great performance boost when some state of tests is shared between the suits. For example embedded db instance - if it's started before each test and switched off after - it could require more time than test execution itself.
If you applied the forking policy from second recommendation to at most once - than all your tests are running in the same JVM and it's possible to share some class instances.
There are some cons with following to this approach - tests aren't independent anymore. Every test should be implemented with keeping in mind that resources are mutated in parallel.
For example Unit tests that counts the final number of rows in table - should be updated to check by the number of rows by condition. In meantime in before[Each/All] should not clear the whole state of shared resource but the parts are related to current test case only. Here is example how to share the embedded cassandra instance between the Specs:
  5. Migrate to async. In case of testing the async services/functionality instead of blocking the test for checking the results use *Async implementations for the Specs: AsyncFlatSpec, AsyncFunSpec etc. More details are here.

  6. Review your before/after[Each/All] methods. Sometimes the execution of those makes major influence into whole test run. If your close the resources - it could be not mandatory to do it - because of short running JVM process is spawned only for test run. If your delete all the data to clean up the state before the tests then think on how it's possible to isolate the tests - via UUID or unique table names etc, this is mandatory step for recommendation #3

  7. Tune the JVM params, test goal in maven is short duration Java process and these JVM args are representing this fact:

-XX:+TieredCompilation -XX:TieredStopAtLevel=1

Type level programming for Streams

Programming at this level is often considered an obscure art with little practical value, but it need not be so.

Conrad Parker (about Type-Level programming in Haskell)

 

There are a lot of domains where events are coming in stream-like way. The best one example are services based on http protocol.

The languages like Java with imperative programming legacy background are providing a sort of abstraction over stream nature of Http services - where stream is converted to branch-like structure and each leaf is handling in-variants for possible inputs. It's became popular to design systems in RMI way.

Regular handler for HTTP or RPC service is looking like:

It means that all the magic with routing HTTP requests to methods are provided by the frameworks.

Of course there is a hidden complexity behind this - like Filter, Handlers etc.

While Reactive and Functional Programming gain some popularity, streaming DSL's gain popularity in solving Flow handling complexity.

Those flows are well self explaining and easier to maintain/change in future.

The most of primitive domain problems can be easy solved via stream of events from Input to to Sink, like:
  In ~> Step1 ~> Step2 ~> StepN ~> Sink. 

Even if your DSL isn't sexy enough to look like math diagram, you can use some tools to visualise flow in nice ways (for example: travesty)

Of course there is always extra complexity like filtering, conditioning, throttling, backpressure etc. Stream like data handling is looking natural and mostly befit the functional programming, where every flow step is a function that accept In type and returns Out type:

  type Step1 = In1 => Out1

Piping the step functions is looking natural solution. Lest assume concrete input types: ShipmentsRequest, TrackRequest, PricingRequest and Ping:

The simplest way to handle those types is operate with Any Type:

Using Any makes possible to pipe (compose) the functions. Unfortunately compiler doesn't help us with any extra checks to make piping safe.
We can use best practices and operated with the marker trait:


Now compiler helps us to find if we have forgotten to handle Ping message:

It's still just a compiler warning - and it's not always possible to escalate it to the error - if you are not allowed to change compiler arguments. It's not always the case that all incoming message could extend some base trait.

There is other way to present input types - via union type. For example Haskell support declarations like this:


In Haskell it's evolved into Data Kinds language extension. Scala is planning to support this on language level as well in Dotty compiler as Union Types.
There is a fork of Scala - Typelevel Scala that aims to bring it early than Dotty. We will use Shapeless library for type level programming solution.
In Scala we can define our input type as a chain of embedded Either types:


Shapeless library implements some syntax sugar for that solution, here is the same code expressed via the Shapeless:

Than Step 1 can be defined as a Poly1 function determined for all unions (coproducts) types from In1:


As a result type of that step is ShipmentsResponse or TrackResponse or PricingResponse or Pong. The handling for all the cases is checked during the compilation time.

The full example for akka streams can look like: