kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From John Roesler <vvcep...@apache.org>
Subject Re: Event Sourcing with Kafka Streams and processing order of a re-entrant pipeline
Date Sun, 31 Jan 2021 20:24:17 GMT
Hi David,

Thank you for the question.

If I can confirm, it looks like the "operations" topic is
the only input to the topology, and the topology reads the
"operations" topic joined with the "account" table and
generates a "movements" stream. It reads (and aggregates)
the "movements" stream to create the "account" table.

I think your concern is dead-on. The processing of incoming
records from either the "operations" topic or the
"movements" topic is synchronous, BUT the production of
messages to the "movements" topic and subsequent consumption
of those messages in the join is asynchronous. In other
words, you can indeed get "insufficient funds" in your

Following your scenario, this can happen:

1. consume Operation(10)
1. join (balance is $0)
1. produce Movement(10)

2. consume Operation(10)
2. join (balance is $0)
2. produce Movement(10)

3. consume Movement(10) // the first one
3. update table to value $10

4. consume Operation(-20)
4. join (balance is $10) // only seen the first one
4. produce Movement(0, "insufficient funds)

5. consume Movement(10) // the second one (too late)
5. update table to value $20

6. consume Movement(0, "insufficient funds)

Other interleavings are also possible.

To get synchronous processing, what you want is a single
subtopology where all the data flows are internal (i.e.,
that re-entrant topic is the source of the race condition).

If you don't know about it already, you can print out your
topology and visualize it with
https://zz85.github.io/kafka-streams-viz/ .

Technically, you can actually model this as a single

 .aggregate(Account(0)){ (_, operation, account) =>
  if (account.balance >= -operation.amount) {
  } else {
   account.copy(error="Insufficient funds")

But if you want to decouple the "insufficient funds" error
handling from the account maintenence, you might look at a
ValueTransformer, in which you maintain the Accounts in a
key/value store and then forward an Either[Account, Error]
result, which you can then direct however you please.

Either way, maintaining the table and the balance checking
logic in a single operation guarantees you won't have race

A final note: the reason your tests don't show the race
condition is that they are using TopologyTestDriver, which
synchronously propagates each individual input record all
the way through the topology. If you also set up a full
integration test, I suspect that you'll quickly see the race
condition surface.

I hope this helps,

On Sun, 2021-01-31 at 11:37 +0100, Davide Icardi wrote:
> I'm working on a project where I want to use Kafka Streams for Event
> Sourcing.
> General idea is that I have a "commands" topic/KStream, an "events"
> topic/KStream and a "snapshots" topic/KTable.
> Snapshots contains the current state of the entities. Commands are
> validated using the "snapshots" and transformed to "events".
>     Group EVENTS stream by key and aggregate them to a SNAPSHOTS table.
>     Left join COMMANDS stream with the SNAPSHOTS table and output new
> For example, to apply this pattern to a simple bank-account scenario I can
> have:
> - operations stream as "commands" (requests to deposit or withdraw an
> amount of money, eg. "deposit $10" => Operation(+10) )
> - movements stream as "events" (actual deposit or withdraw event, eg. "$10
> deposited" => Movement(+10) )
> - account table as a "snapshots" (account balance, eg. "$20 in account
> balance" => Account(20) )
> - account id is used as key for all topics and tables
> The topology can be written like:
>     case class Operation(amount: Int)
>     case class Movement(amount: Int, error: String = "")
>     case class Account(balance: Int)
>     // events
>     val movementsStream = streamBuilder.stream[String,
> Movement](Config.topicMovements)
>     // snapshots
>     val accountTable = movementsStream
>       .groupByKey
>       .aggregate(Account(0)){ (_, movement, account) =>
>         account.copy(balance = account.balance + movement.amount)
>       }
>     accountTable.toStream.to(Config.topicAccounts)
>     // commands
>     val operationsStream = streamBuilder.stream[String,
> Operation](Config.topicOperations)
>     operationsStream
>       .leftJoin(accountTable) { (operation, accountOrNull) =>
>         val account = Option(accountOrNull).getOrElse(Account(0))
>         if (account.balance >= -operation.amount) {
>           Movement(operation.amount)
>         } else {
>           Movement(0, error = "insufficient funds")
>         }
>       }
>       .to(Config.topicMovements)
> (see full code here:
> https://github.com/davideicardi/es4kafka/blob/master/examples/bank-account/src/main/scala/bank/StreamingPipeline.scala
> )
> Now let's imagine a scenario where I deposit $ 10, then I deposit again $
> 10 and then I withdraw $ 20:
>         inOperations.pipeInput("alice", Operation(10))
>         inOperations.pipeInput("alice", Operation(10))
>         inOperations.pipeInput("alice", Operation(-20))
> Can I assume that when processing the third message (withdraw $ 20) the
> account table already processed the previous two movements (10+10)?
> In other words, can I assume that:
> - if operations are valid I never receive an "insufficent funds" error
> event ?
> - in the above topology, account KTable is always updated before processing
> the next operation from KStream ?
> From my tests it seems to work, but I would like to have some advice if
> this is a safe assumption.
> (see test here:
> https://github.com/davideicardi/es4kafka/blob/master/examples/bank-account/src/test/scala/bank/StreamingPipelineSpec.scala
> )
> thanks
> Davide Icardi

View raw message