0

I have a Kafka topic providing events of the following type: id(hash):[ADD|REMOVE].

  1. These events may be generated at a high rate and are idempotent, i.e. getting 123:ADD one time and ten times in a row produces the same result: state of 123 is ADDED
  2. Processing of each event may take a while (not hours, but seconds).
  3. Processing of each event may generate new events for different keys. Ex: 123:ADD may produce secondary event 234:REMOVE or any other number of key/action events.
  4. For same keys secondary event should be overridden by primary event if its parent was received before the overriding event.
  5. Model state should represent result of the last event received at the moment.

Amount of events is significant, so in a linear single-threaded way this is too slow. Multiple processor instances are expected.

While I can use Kafka partitioning by key to assure primary events are processed in the right order, I'm not sure how to handle the appearing secondary events correctly.

For example:

Timeline Event
1 ms primary 123:ADD
50 ms primary 234:ADD
1100 ms secondary 234:REMOVE is produced while processing the first event.

In this case final state of 234 would be REMOVED however as the parent of secondary event was received before primary 234:ADD the order of event processing should be different and the final state should be 234:ADDED

I thought about adding timestamps to events and making secondary events get same timestamp as their parents. This way it would be possible to discard outdated events.

My questions are: how do I sync all the event processors across multiple instances of a service to guarantee that event with an earlier timestamp is not executed after event with a later timestamp?
How do I guarantee that timestamps are in sync across instances of service?

I might push all the events to some db, but making sure that no event with given key exists at the moment would require a full table lock which to me looks as no good for concurrent access.

So I'd be grateful for any hints or ideas or proving me wrong.


Solution I'm thinking of at the moment:

  1. All the primary events I'm processing are ordered in kafka, so it would be fair to produce these events with a sequentially incrementing timestamp or id. Second option is to use Kafka offset to order the events.
  2. I can add parent's timestamp/offset to spawned events as I know exactly who spawned it.
  3. After processing these events I can propagate them to a different kafka topic partitioning by key.
  4. I can add another service consuming data from second topic and adding pairs key/timestamp to a database.
  5. If the event I'm consuming has an earlier timestamp, drop it, otherwise update timestamp in db and perform action.

Partitioning in Kafka guarantees all events with same key are handled by the same consumer and this way may be processed in linear mode. At the same time events with different keys may be processed in parallel. Clock sync shouldn't be an issue in this case as all the timestamps/offsets are set by an ordered primary producer.

This way there even may be no need in table locks as concurrent access to same data shouldn't happen by design.

2
  • Is it possible to add a transaction Id or sequence number to each event? That would get tricky for events that spawn new events, though. Dec 12, 2022 at 22:31
  • They are ordered already in kafka. It is the spawned events that break the concept.
    – svz
    Dec 13, 2022 at 6:33

2 Answers 2

2

This is addressed in Leslie Lamport's 1978 paper, "Time, Clocks, and the Ordering of Events in a Distributed System". The short answer to your questions are:

  1. You can enforce an "happens-before" ordering on all events in a distributed system.

  2. You cannot totally accurately synchronize all clocks in a distributed system.

The topic to do some more research into would be "Logical Clocks," which are used precisely in this situation where no physically synchronous clock exists for the entire system. There are several different flavors with increasing complexity and advantages.

That all said, in the real world there are often additional constraints that can make the problem easier or harder. For example, clock drift is more of a problem the more accurate timestamps you want, or the "laggier" your distributed network becomes. Parallel processing cores might have clock drift problems on the order of nanoseconds or microseconds, while networked distributed machines might have clock drift on the order of milliseconds to seconds.

In many small-scale distributed systems, it is sufficient just to have machines synchronize with a specific ground truth server occasionally, and then just trust the timestamps you get.

What is the failure consequence? If you're building a videogame and the failure consequence is that someone misses an opportunity then you're probably OK with this approach. If you're trying to create a legally binding ordering on financial transactions then you have more legal problems then technical problems. For example, stock exchanges address this problem with legal language saying that transaction requests are only ever handled on a best-effort basis and no specific guarantee of event handling is provided. Banks for another example will reserve the right to reorder all transactions received on a checking account within a given day.

The types of events you handle matter as well. For example, addition and subtraction are commutative, so the order you handle those events don't actually affect the end-of-day totals, if that end-of-day total is what you care about. Addition and multiplication are not commutative, so order really matters.

1
  • The events I'm handling are not commutative. They define state of a system in terms of on/off, so change of ordering will lead to a totally different result. At the moment my question has shrinked to 'how do I effectively implement a high performance sync of multiple processor instances across a single db table?'.
    – svz
    Dec 13, 2022 at 6:29
2

Perhaps the expensive processing could be parallel, but the result of that processing could be added to a thread-safe priority queue and reported or persisted by a single thread.

But first I would question the assumption that events must be processed in the order of their timestamps. The question of whether the timestamps originate on a single machine or multiple machines with potentially different clocks is only the beginning. In highly parallel systems, don't focus on the order that events actually occurred. Instead, ask whether it would have been wrong for them to occur in a different order, had thread scheduling or network latency been slightly different. If not, don't worry about it!

1
  • Yes, order of processing does matter. In fact, cases like the one I described in the example are quite rare - few in a million, maybe, but it is critical not to allow a single case like that.
    – svz
    Dec 13, 2022 at 6:24

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.