Lagom: Offset tracking grouping

[jroper]

Currently Lagom's APIs that do offset tracking (read side processors and topic publishers) persist the offset of every message. This is not a bad default, especially for read side processors, as it allows exactly once processing. But exactly once processing isn't always necessary, and the performance cost could be high. So we should offer a configuration option to specify grouping, so that, for example, the offset can be configured to only be persisted every 10 events.

[TimMoore]

A related idea would be batching the actual effects of the read-side processor.

In Cassandra this would look like aggregating lists of statements from multiple events into a single batch update.

In JDBC, this would be wrapping a transaction around multiple calls to the event handler.

This would preserve exactly-once processing while also allowing for more efficient use of the database.

[jroper]

Good point.

There is a danger in Cassandra if effects are dependent on reads, as the reads will be done before anything in the batch is executed, and so if there are any dependencies between events in the batch, those dependencies may not be realised. Incrementing a counter would be a good example of this. So I think both should potentially be offered.

In a relational database though transactions should see their own uncommitted writes, so it shouldn't be a problem.

[lutzh]

How would one define when to write? If you configure a fixed batch size, that might delay the update of the read side significantly (i.e. if you say you write batches of 10, eight events occur, but it takes an hour for the next two events to occur, the read side update would be delayed for an hour, or am I misunderstanding?).
So it would be preferable to do it time-based, with a time-based "sliding window", do you agree?

[lutzh]

@jroper Question regarding the initial statement - does batching the write of data and offset really guarantee exactly-once-processing for Cassandra read sides? I was under the impression there could still be conditions that lead to to partially applied batches, e.g. if the Cassandra coordinator node fails during the batch execution. Does Lagom guarantee exactly-once in the Cassandra read side processor?
My assumption was read side updates on Cassandra are "at-least-once" and may be reapplied in case of failure during updates (i.e. in a scenario where the data has been written, but the offset hasn't). Was I too cautious?

[ignasi35]

How would one define when to write? If you configure a fixed batch size, that might delay the update of the read side significantly (i.e. if you say you write batches of 10, eight events occur, but it takes an hour for the next two events to occur, the read side update would be delayed for an hour, or am I misunderstanding?).

Maybe a combination of both, e.g. transaction is committed every 10 events or after 10 seconds since it started, whichever happens first., would do the trick.

Overall, this will add an extra delay on the read-side (already defaults to a 10sec delay).

[jroper]

We'd use Akka streams groupedWithin, which groups a maximum number of events within a given time frame. The timeout would be far lower than 10 seconds - batching really only matters when high throughputs (starting at hundreds per second) need to be achieved.

As to whether Lagom guarantees exactly once in Cassandra read side processing, the answer is yes and no. It depends on what your read side processor does. If your processor doesn't directly do any updates, but just returns it's update statement(s) to Lagom, then Lagom will add the offset tracking update statement to that list, and execute them in a Cassandra batch. Cassandra batches are atomic operations, either all the update statements succeed, or they all fail. This yields exactly once semantics.

The same goes for Lagom's JDBC (and JPA and Slick) read side processors, since in those cases the offset storage is done in the same transaction as the update, making it atomic, and therefore yielding exactly once semantics.

[aklikic]

+1