Message Group / Sequential Convoy pattern

[stepin]

I need to process messages in parallel but in some cases, they should be processed in sequence.
A good simple example is the order case: messages for different order ids should be processed in parallel
but for the same should be processed in sequence. This pattern is described in more details at Microsoft
https://docs.microsoft.com/en-us/azure/architecture/patterns/sequential-convoy
and at ActiveMQ http://activemq.apache.org/message-groups.html . It's a common use case for Event Sourcing scenarios.

Requirements:

• at least once delivery (and exactly one within a window is even better)
• messages have a sequence key: if it's null no limitations on parallelization, if it's not null its name of a sequence
• single workers' group (in worst case 2 groups for null and non-null sequence keys)
• timeouts for processing
• up to 5 attempts to process (in error case it's better to send failed messages into the dedicated topic)

I see several options:

1. NATS JetStream Workers

• NATS JetStream Workers for jobs that can run in parallel
• messages with ID header for exactly once detection
• all preprocessing in a custom code with external storage for sequences

In this case, it's unclear how my code will receive notification that
some message was processed (to check if there are more messages to
process in the same sequence).

2. NATS Core Services

• NATS Core Service for jobs that can run in parallel
• all preprocessing in a custom code with external storage for all messages in progress

It's clear how to implement this option but too many queue-related code outside of the queue.

Is this possible to implement the original pattern with storage inside JetStream? Or maybe even without custom code
(maybe I missed or misunderstood some options like MaxConsumers)? Is this pattern planned for implementation in the future?

[ripienaar]

We've discussed this one a bunch, I think we will need a solution to this but do not have an elegant one today - almost there but we have no way to rebalance consumers.

One for the back burner for future releases I think

[ripienaar]

See nats-io/jetstream#103

[stepin]

Ok, I will close this as duplicate.

[cortopy]

could this be reopened?

nats-io/jetstream#103 and the repo itself were archived because jetstream is now in GA and part of the nats server, which leads back to here again?

[derekcollison]

We have been discussing this one quite a bit. No resolutions yet but it's definitely on our radar and part of active discussions.

[timsolov]

Hi, I have an idea. What will say community?

We can use distributed atomic lock system (for example ZooKeeper or ETCD).

1. Consumer pulls 20 messages.
2. Consumer iterates over messages' payloads.
3. Consumer looks to some special key inside each payload (e.g. order_id) and tries to set lock in zookeeper say in /orders/{order_id}.
4. If lock already exists then we skip this message by sending msg.Nak() (not acknowledged) because that means a message with same key already handling by another consumer.
5. If lock doesn't exist then we'll take it and go to handle the message.

This logic can be implemented also inside the nats-server. But I don't know how this affects on performance.

[ripienaar]

JetStream already does this. When you pull 20 messages you have a lock on that message for AckWait time.

Is it that you want to arrange it so that one consumer always handle all messages related to a given order ID? If that’s what you are asking then we are looking at adding something to the server itself. Currently discussed here nats-io/nats-architecture-and-design#36

[timsolov]

@ripienaar not really, when we have GroupSubscription and 3 parallel consumers we can receive different events simultaneously which in microservice arch should be handled one by one.
Example:

1. Customer creates invitation and removes this invitation straight off.
2. First consumer starts handle creation of invitation but doesn't have time complete transaction when another event (remove invitation) comes to Second consumer. And the Second consumer tries to remove non-existing invitation.

So what the nats-server offers to handle this situation?

There are two ways as I see to resolve this:

1. Like a Kafka: send messages by Key from header to different partitions.
2. Something what I've suggested before in Message Group / Sequential Convoy pattern #2043 (comment)

[ripienaar]

Yes, this is the feature we are working on fleshing out in the link given nats-io/nats-architecture-and-design#36

[rbkrabbe]

Any news on the feature design? I am also looking for this capability.

[derekcollison]

We have some updates in 2.8 for folks that are more aligned to the way other tech does partitioning. @jnmoyne might be able to give some insight.

We are still working on a the solution though described above.