Add batch direct get #1229

MauriceVanVeen · 2024-09-18T15:21:12Z

Implements batch direct get as defined in ADR-31.

MauriceVanVeen · 2024-09-18T15:23:15Z

src/main/java/io/nats/client/impl/NatsJetStreamManagement.java

+
+        CachedStreamInfo csi = getCachedStreamInfo(streamName);
+        if (!csi.allowDirect) {
+            throw new IllegalArgumentException("Allow direct not enabled for stream: " + streamName);


There are two non-server error conditions:

either AllowDirect is not enabled, in which case we'd get a timeout from the server

the server is not 2.11+ and doesn't support this feature yet

Currently made both of them IllegalArgumentException, but not sure if that's the right fit..
JetStreamApiException doesn't fit either as it's for server errors.

I think this is fine. Are you going to update NatsJetStreamImpl feature flags to know if the server supports? See lines 48/49 and 71/72. What's a good name for this feature? "direct211"?

Are you going to update NatsJetStreamImpl feature flags to know if the server supports?

Will do 👍
Wondering whether those fields will be updated if we were initially connected to a server that supports it, and we reconnect to another server that doesn't?

What's a good name for this feature? "direct211"?

Yeah, something like that. directBatchGet211Available, maybe

Added as directBatchGet211Available.
And added exceptions for these cases: JsDirectBatchGet211NotAvailable and JsAllowDirectRequired.

Signed-off-by: Maurice van Veen <[email protected]>

scottf · 2024-09-23T16:28:01Z

src/main/java/io/nats/client/JetStreamManagement.java

+     *         server such as timeout or interruption
+     * @throws JetStreamApiException the request had an error related to the data
+     */
+    void getMessageBatch(String streamName, MessageBatchGetRequest messageBatchGetRequest, Consumer<MessageInfo> consumer) throws IOException, JetStreamApiException;


Don't use "Consumer". You should make something like MessageInfoHandler, similar to MessageHandler. The justification for this is that we have a class called Consumer and I don't want there to be any confusion, especially when it comes to user's code and their imports. This is not the first time I've made this choice.

Done, added MessageInfoHandler.

scottf · 2024-09-23T16:33:10Z

src/main/java/io/nats/client/api/MessageInfo.java

        }
        JsonUtils.addField(sb, TIME, time);
        JsonUtils.addField(sb, STREAM, stream);
        JsonUtils.addField(sb, "last_seq", lastSeq);


Just noticed there is an ApiConstant.LAST_SEQ. Can you update this since you are here.

Done, using LAST_SEQ and also ERROR above existed already.

scottf · 2024-09-23T16:35:40Z

src/main/java/io/nats/client/support/NatsJetStreamClientError.java

    public static final NatsJetStreamClientError JsConsumerNameDurableMismatch = new NatsJetStreamClientError(CON, 90302, "Name must match durable if both are supplied.");
    public static final NatsJetStreamClientError JsMultipleFilterSubjects210NotAvailable = new NatsJetStreamClientError(CON, 90303, "Multiple filter subjects not available until server version 2.10.0.");
+    public static final NatsJetStreamClientError JsAllowDirectRequired = new NatsJetStreamClientError(CON, 90304, "Stream must have allow direct set.");
+    public static final NatsJetStreamClientError JsDirectBatchGet211NotAvailable = new NatsJetStreamClientError(CON, 90305, "Batch direct get not available until server version 2.11.0.");


Please update the README

src/main/java/io/nats/client/impl/NatsJetStreamManagement.java

scottf · 2024-09-23T17:24:06Z

src/main/java/io/nats/client/impl/NatsJetStreamManagement.java

+            long maxTimeMillis = messageBatchGetRequest.getTimeout().toMillis();
+            long timeLeft = maxTimeMillis;
+            while (true) {
+                Message msg = responseRequired(sub.nextMessage(timeLeft));


Let's talk through a few things.

The user processing the message takes time away from your wait.

Do we //how would we notify the user about when we've asked for 10 messages, the server has plenty in the range, but there is a failure after less than 10, like a real disconnection.

Are there other ways to give the user messages, meaning should they have the choice to fetch, iterate or consume (callback is consume)

Signed-off-by: Maurice van Veen <[email protected]>

scottf

LGTM

scottf

Changes discussed

Signed-off-by: Maurice van Veen <[email protected]>

scottf

Naming nitpick

scottf · 2024-10-02T17:05:28Z

src/main/java/io/nats/client/JetStreamManagement.java

+     *         server such as timeout or interruption
+     * @throws JetStreamApiException the request had an error related to the data
+     */
+    void gatherMessageBatch(String streamName, MessageBatchGetRequest messageBatchGetRequest, MessageInfoHandler handler) throws IOException, JetStreamApiException;


I got rid of "gather" and just did "request" so as not to make it seem like I only support "scatter gather". I also changed "iterate" to "queue" since it seems more accurate and also since we aren't giving them an iterator.

Signed-off-by: Maurice van Veen <[email protected]>

scottf

LGTM

MauriceVanVeen commented Sep 18, 2024

View reviewed changes

Add batch direct get

00a8a02

Signed-off-by: Maurice van Veen <[email protected]>

MauriceVanVeen force-pushed the batch-direct-get branch from c3dac04 to 00a8a02 Compare September 23, 2024 15:25

MauriceVanVeen marked this pull request as ready for review September 23, 2024 15:47

MauriceVanVeen requested a review from scottf September 23, 2024 15:47

Remove MessageBatchInfo

c93fa00

Signed-off-by: Maurice van Veen <[email protected]>

scottf requested changes Sep 23, 2024

View reviewed changes

scottf and others added 5 commits September 23, 2024 14:05

Merge branch 'main' into batch-direct-get

e6f9a4f

Use pre-existing ApiConstants

7c3f4ea

Signed-off-by: Maurice van Veen <[email protected]>

Add MessageInfoHandler

63f3fce

Signed-off-by: Maurice van Veen <[email protected]>

Add new errors to README

ca5e800

Signed-off-by: Maurice van Veen <[email protected]>

Update method signature to return queue

cda1e2e

Signed-off-by: Maurice van Veen <[email protected]>

scottf approved these changes Sep 24, 2024

View reviewed changes

scottf requested changes Sep 24, 2024

View reviewed changes

Support fetch/iterate/consume semantics & add EOD

a241775

Signed-off-by: Maurice van Veen <[email protected]>

MauriceVanVeen requested a review from scottf September 25, 2024 14:14

Rename consume to gather

0e498c2

Signed-off-by: Maurice van Veen <[email protected]>

scottf requested changes Oct 2, 2024

View reviewed changes

MauriceVanVeen added 2 commits October 3, 2024 09:33

Rename to request and queue

b5c4c0b

Signed-off-by: Maurice van Veen <[email protected]>

Add error handling test for fetch and queue

38f3c53

Signed-off-by: Maurice van Veen <[email protected]>

scottf approved these changes Oct 3, 2024

View reviewed changes

scottf merged commit 27efeab into main Oct 3, 2024
4 checks passed

scottf deleted the batch-direct-get branch October 3, 2024 10:19

Gerschtli mentioned this pull request Jun 13, 2025

Add support for Direct Get Batch again #1330

Closed

Add batch direct get #1229

Add batch direct get #1229

Uh oh!

Conversation

MauriceVanVeen commented Sep 18, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

scottf left a comment

Choose a reason for hiding this comment

Uh oh!

scottf left a comment

Choose a reason for hiding this comment

Uh oh!

scottf left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

scottf left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!