Keeping up logical order in a messaging environment

Question

My setting is the following (Please assume that this points are entirely rock solid unchangeable, some for good reasons some: "just because"):

• In a super scalable microservice environment i receive messages
• These message have a logical order (a timestamp that they bring with them)
• The messages are fed into a messaging system
• The messages are idempotent
• As soon as a message has arrived in the data storage it is okay to discard any older message for the same objectId
• Neither the technical order of the messages (the order of the publishing to the system) can be guaranteed nor its uniqueness (messages can be delivered multiple times)
• Consumers are consuming messages in a highly parallel fashion
• At the end of the processing of the messages most of them will be stored in a SQL Database
• No hard consistency constraints apply (eventual consistency can be applied)
• Using Kafka order other event streaming platforms is too expensive because of its total cost of ownership

So the central problem i am facing is: What to do if a older messages gets processed after a newer one.

Here is a simple example, lets say the following messages are processed in this order:

{
  objectId: '4711'
  logicalOrder: 2
  content: "World"
}

{
  objectId: '4711'
  logicalOrder: 1
  content: "Hello"
}

In the end the field content will be written to the database. So if i do nothing special about it, the result will be wrong:

+----------+---------+
| objectId | content |
+----------+---------+
|     4711 | Hello   |
+----------+---------+

Now the first thing that comes to mind is adding the logicalOrder as a column and then there are two approaches.

1. Adding everything that is processed to the table (keeping up the ability to easily batch insert records)
2. Checking if something newer (according to logicalOrder) is in the table and only write to the table if there is no newer dataset in the table

In the first case we will end up with this:

+----------+---------+--------------+
| objectId | content | logicalOrder |
+----------+---------+--------------+
|     4711 | World   |            2 |
|     4711 | Hello   |            1 |
+----------+---------+--------------+

Now here comes the questions:

• What do you think is more performant in a SQL environment? Taking the overhead of checking first before writing to the table or
• using a PARTITION BY fancy sql statement to always select the entries with the highest logicalOrder evertime data is read (maybe also a cleanup job will be needed to free storage and to bring back performance).
• Do you encounter similar problems?
• How did you solve them?

Answer 1

It sounds like you are inventing Event Sourcing.

I would avoid any processing on the SQL side, have my code load all the operations and process them in logical order into build the object state. This keeps your inserts fast and you can occasionally do snap shotting to limit the amount of data needed.

However, you also have the problem of the time after event 2 has been processed but event 1 hasn't been seen yet.

Your code can recognise this state and take appropriate action if you include some sort of event ordinal which allows you to check for missed messages.

Answer 2

I don't think you can easily solve this problem just by using clever SQL. One pattern that you may find useful here is the Actor Model, which you can combine with Consistent Hashing to build a system that effectively manages both parallelism and logical in-order processing of related messages.

Basically, you can use the actor model to create agents that process messages one-at-a-time in order. The actors can use a priority queue as a mailbox and order the messages by the field that determines their logical ordering. The mailbox can also keep track of the greatest value for a processed message and report errors or discard messages that arrive too late to be processed in-order.

You can then create "Router" actors that don't process messages themselves, but instead just distribute messages to other actors. These actors can use a consistent-hashing algorithm to route all messages for a given topic (say, an objectId) to the same actor, so that they are sorted in the correct way by the priority-queue mailbox, but also process messages for unrelated topics in parallel.

If you're using .NET or Java platforms, you can look at the Akka or Akka.NET frameworks to get all of these tools from an out-of-the-box solution. Here's an article about this sort of design for Akka.NET: https://petabridge.com/blog/akkacluster-state-distribution/

Answer 3

I personally would not do this in a way that requires any back-and-forth messaging between the application and database. Leave the work to the RDBMS.

You want to insert if the data does not exist else update if it is stale. I think the best option to achieve this is to essentially do an "UPSERT".

You might look into how to do this on your particular RDBMS but for SQLServer it can be done using a MERGE:

Create Table YOUR_OBJECT (ID integer, LOGICAL_ORDER integer, DATA varchar(100))

Merge YOUR_OBJECT as trg
 Using (Select ? as ID, ? as LOGICAL_ORDER, ? as DATA) as src
 On trg.ID = src.ID
 When Matched And trg.LOGICAL_ORDER <= src.LOGICAL_ORDER Then
  Update Set trg.LOGICAL_ORDER = src.LOGICAL_ORDER, trg.DATA = src.DATA
 When Not Matched Then
  Insert (ID, LOGICAL_ORDER, DATA) Values (src.ID, src.LOGICAL_ORDER, src.DATA);

Then you can use prepared statements and set parameters 1, 2, and 3, as the ID, LOGICAL_ORDER, and DATA respectively to possibly insert or update.

Here is a small performance comparison of the MERGE command

Here's an example for Java:

public static void main(String[] args) {
    StringBuilder upsert = new StringBuilder();
    upsert.append("Merge YOUR_OBJECT as trg");
    upsert.append(" Using (Select ? as ID, ? as LOGICAL_ORDER, ? as DATA) as src");
    upsert.append(" On trg.ID = src.ID");
    upsert.append(" When Matched And trg.LOGICAL_ORDER <= src.LOGICAL_ORDER Then");
    upsert.append("  Update Set trg.LOGICAL_ORDER = src.LOGICAL_ORDER, trg.DATA = src.DATA");
    upsert.append(" When Not Matched Then");
    upsert.append("  Insert (ID, LOGICAL_ORDER, DATA) Values (src.ID, src.LOGICAL_ORDER, src.DATA);");
    try (Connection con = ...) {
        ...create table...

        final int ID = 1;
        final int LOGICAL_ORDER = 2;
        final int DATA = 3;
        try (PreparedStatement ps = con.prepareStatement(upsert.toString())) {
            // call 1
            ps.setInt(ID, 0);
            ps.setInt(LOGICAL_ORDER, 2);
            ps.setString(DATA, "World");
            ps.execute();

            // call 2
            ps.setInt(ID, 0);
            ps.setInt(LOGICAL_ORDER, 1);
            ps.setString(DATA, "Hello");
            ps.execute();
        }
        con.commit();
    } catch (SQLException e) {
        e.printStackTrace();
    }
}