Add bulk message ack/nak

[matthewadams]

Proposed change

NB: I asked this question in the #general channel of the Slack channel, but didn't seem to get much traction. The proposed methods are slightly different than in the Slack messages after some more thought.

For those use cases where we don't know at design time which AckPolicy we want to use, we'd like for there to be more efficient bulk acking/naking APIs on FetchConsumer. These are intended primarily for ordered pull consumers. We propose the following new interface methods.

• default void ackAll() { ackAll(true); }
• default void nakAll() { ackAll(false); }
• default void ackAll(boolean ack) { ack(-1, ack); } -- acks all messages in one server round trip, similar to AckPolicy.All
• void ack(int n, boolean ack) { ack(n, ack, false); } -- acks first n messages (all if n < 0) of the batch in one server round trip
• void ackThenNak(int n); -- acks first n messages (all if n < 0) then naks the rest in one server round trip

If it's inappropriate for these methods to be placed on FetchConsumer, perhaps a new set of batch APIs could be created on ConsumerContext to support batches of Messages, like

• BatchConsumer fetchBatch(int maxMessages);
• BatchConsumer fetchBatch(int maxBytes);
• BatchConsumer fetchBatch(BatchConsumeOptions opts);

where BatchConsumer might be something like

package io.nats.client;

public interface BatchConsumer {
    List<Message> getBatch() throws InterruptedException, JetStreamStatusCheckedException;
    int getBatchSize();
    default void ackAll() { ackAll(true); }
    default void nakAll() { ackAll(false); }
    default void ackAll(boolean ack) { ack(getBatchSize(), ack); } 
    void ack(int n, boolean ack);
    void ackThenNak(int n);
}

Use case

The use case is when it's not known at design time which AckPolicy should be used, and the developer doesn't want to ack or nak each Message individually.

[Jarema]

Hey!

Thanks for the proposal and effort to write it down.

Let's address few points:

1. Ordered Consumers always use AckNone. That is a base principle for their mechanics to make them work.
2. The concept of acking / nacking all` is tricky in Pull Consumers. If you have multiple clients fetching messages from the same Consumer, you could ack someone else's message.
3. Concept of single roundtrip: Ack is just a publish. Many acks can actually happen in single roundtrip to the server. NATS client tries to efficiently buffer messages before sending them out without sacrificing latency.
4. Introducing such API without server changes would be misleading - as each ACK is a separate publish to separate subject, something could break in the middle resulting in some messages being acked, some not. Single roundtrip or not - it would still not be an atomic operation.

Because of above, I do not think its a good idea to implement what you are suggesting.
You can of course use some thin abstraction layer on top of current client API, but I wonder if above points do still make it valid even for your specific use case?

Feel encouraged to discuss it further.

[ripienaar]

I've converted this to a discussion.

[matthewadams]

Ok. Perhaps this is an opportunity to improve the stream consumer documentation and/or the various clients' JetStream APIs, because my experience configuring JetStream consumers (using jnats for me) has been confusing and difficult since starting to use NATS, despite the newer, simpler stream consumer API. Don't get me wrong, I'm a huge NATS fan; I just dread trying to configure stream consumers at this point.

It appears to me that there are only so many factors when it comes to consumers:

• push or pull,
• ephemeral or durable, and
• ordered or unordered.

I'm also wondering how the type of stream affects these choices as well; I'm referring specifically to streams with the three different RetentionPolicys.

It surprises me that you assert that ordered consumers inherently use AckNone. If you could point me to documentation that explains this, it'd help.

My use case is that I want to primarily use pull consumers to implement CQRS event projectors that are receiving events from a stream that represents an event bus. Some projector types are simply writing the events themselves to a database, some are updating counts, averages, etc to projections, and some are calling third-party partners' systems, etc. These are the features that I've identified for these three types of projectors.

• Projector type simply storing events in a database: unordered pull consumer, unsure about durability.
• Projector type updating moving averages, etc: ordered pull consumer, unsure about durability.
• Projector type integrating with third-party systems: ordered pull consumer, unsure about durability.

As you can see, I'm unsure about durability. We're operating in a clustered environment, so that we have multiple instances of identical projector types at any given time available to process batches of messages. In our projector use cases, we don't want to miss any messages, so I conclude that they should all be durable. If I'm incorrect, please explain why.

Also, if a projector type pulls a batch of, say, 100 messages, successfully processes 63 of them and then encounters an error, I expected to ack the first 63 messages so that they're not redelivered to any other projector instances of the same type, but if you're saying that messages are not acked, then I don't understand.

Perhaps the docs should enumerate each of these nine permutations of push/pull, ephemeral/durable, and ordered/unordered individually and describe how they should be configured and whether they're nonsensical. Further, I'd think it pretty simple if the client APIs had simpler, high-level APIs for each of these permutations to simplify writing consumers. In the case of jnats, for example, methods on StreamContext could be added like

• createOrderedDurablePullConsumer(/* options exclusive to this consumer type */);
• createUnorderedDurablePullConsumer(/* options exclusive to this consumer type */);
• createOrderedEphemeralPullConsumer(/* options exclusive to this consumer type */);
• etc

Thanks in advance for setting me straight.

NB: while the video at https://docs.nats.io/nats-concepts/jetstream/consumers#persistence-durable-ephemeral was informative, I guess it wasn't enough for me to master consumer configurations.

[matthewadams]

After reading https://docs.nats.io/nats-concepts/jetstream/consumers/example_configuration, I'm even more confused. The paragraph below the image states:

These operations are all pull based meaning they are work queues and can scale horizontally. All require acknowledged delivery ensuring no order is missed.

That statement, even in context, appears to imply that messages need to be acknowledged even though the consumers are pull-based.