Max Ack Pending per Subject?

[praveenperera]

I have a consumer that listens to a stream on subjects client.node.*

Each node sends messages to the subject with its node id. So node id a would be client.node.a

I would like to set max ack pending for each node’s messages to 1 so I can process the messages of each individual node in order.

But I want to process nodes a,b and c’s messages in parallel to each other. Hope that makes sense.

Can I do that with one consumer? Or do I need to make a different consumer for each node?

[Jarema]

Right now it is unfortunately not possible to set max ack pending per subject.
In your use case, you need to create consumer per node, as having max ack pending above 1 can cause out of order redeliveries on not acked messages (missed acks, network issues, etc).

[john9x]

@jnmoyne Thanks! I see this but I need Java implementation and I don't understand how to use it in my case. Partitioned Consumer Group per device/subject?

[jnmoyne]

It all depends on whether you need 'strictly ordered (per device) processing of the messages' or not. If you don't you could have a pool of worker threads in your client application set max acks pending to > 1 and have one thread getting messages from the consumer and handing them off to worker threads for processing, that means you can parallelize the processing of those messages such that one message being slow to process doesn't affect other messages that are fast to process.

Now if you need to process those messages in strict 'per device' order, then you need to set max acks pending to 1 and that severely limits your throughput as in that case there is no parallelization of the work possible. In that case, in order to regain that parallelization and keep the strict per device ordering you can use the PCG library to have multiple instances of your client application running in parallel and each one consuming on a subset of the devices (with max acks pending 1), such that processing a slow message on one of the partitions doesn't affect the processing of messages on other partitions.

[john9x]

@jnmoyne Thanks! I need 'strictly ordered (per device) processing of the messages'.
Now we use Kafka and already have parallelization by partitions -> processing a slow message on one of the partitions doesn't affect the processing of messages on other partitions but It is not an ideal. Partitions in the Kafka is not a cheap (you know I'm sure) so one device affects many other in the same partition. That is why I'm start to find better solution and thinking about NATS.

[jnmoyne]

PCG is (as the name implies) much like consumer groups and partitions in Kafka, but 'partitions' in this case are just part of the subject for the messages in the stream but still they are not 'infinite' either.

I think in your case I would do the following: have a fixed size pool of threads for processing messages in your client app, create a consumer with max acks pending to some value > 1 and at least more than the number of those threads. Then you have the callback from the Consume on the consumer which will get a message from the stream, extract the device id, hash it with a modulo of the number of threads and then send it for processing to the thread it got assigned to, it's like doing partitioning with threads within your client application. And then if you need more than one instance of your application running then you use a PCG to scale to multiple instances, still using partitioning but doing this two-level partitioning you may be able to get to a large number of 'partitions'.

[john9x]

get a message from the stream, extract the device id, hash it with a modulo of the number of threads and then send it for processing to the thread it got assigned to, it's like doing partitioning with threads within your client application.

Yeah! It is already implemented exactly as you wrote on top of Kafka! :)

But still I don't see an implementation of PCG in the nats-java client or I miss something?