Jetstream partitioning without impact on consumers

[OpenGuidou]

Hello,

I have a high-throughput use case with the following architecture:
Streams definitions:

Stream 1:

• subject test1.>

Stream 2:

• subject test2.>

Message flow:

Message on "test1.>"   
  ->  consumed by App-A 
       -> producing on subject test2.1.2.3 
            -> consumed by App-B on test2.1.>
            -> consumed by App-C on test2.>

There is one jetstream subscription per App-B instance, to test2.1.>
There is one jetstream subscription per App-C instance, to test2.>

App-B and App-C should be able to consume past events, as they can be offline for some time (explaining the need for jetstream).

With a 3 nats nodes cluster, I noticed a very high activity on one of the nodes, and very few activity for the others.
The consumption latency for App-B and App-C (after being produced from App-A) started to degrade as well.

Reading the doc, one idea could be to use the subject mapping to distribute the load on several server nodes.
With a setup like;

Mapping:

• test2...* : test2.$1.$2.$3.{{partition(3,1)}}

Stream 1:

• subject test1.>

Stream 2:

• subject test2...*.1

Stream 3:

• subject test2...*.2

Stream 4:

• subject test2...*.3

I'd still like my App-B and App-C to receive all the messages from Stream 2, 3 and 4 transparently, by using the same subjects, but it doesn't seem to work ("No matching streams for subject." when creating the consumer).

Subscriptions are created this way:

Subscription sub = nc.jetStream().subscribe("test2.>", dispatcher, messageHandler, false,  
      PushSubscribeOptions.builder().deliverGroup("App-C").durable("App-C")).build());

(Taking example from https://github.com/nats-io/nats.java/blob/main/src/examples/java/io/nats/examples/jetstream/NatsJsPushSubQueueDurable.java)

Did I get all of that correctly ? Is there a way to achieve such a scenario ?

[scottf]

If you are getting No matching streams for subject I'm guessing the mapping has an issue. I wonder about the test2...* specifically and what the effect of that is I think it may be literal, not wild carded like test2.*.*.*. Can you post your server config? Also the actual stream config, either from nats s info or from the java management get stream info or the stream config code used to create the stream if you did it from Java.
I'll need to ask about what the exact behavior of the partitioning should be.

[scottf]

@ripienaar any idea about the cli issue?

~ # nats s info
panic: runtime error: slice bounds out of range [:-2]

[OpenGuidou]

In that case, could the "io.nats.client.impl.NatsJetStream#lookupStreamBySubject" be made accessible through the jetstream lib API, so that we can create the subscriptions dynamically based on the results ?
If we want the partitioning to be transparent on client side, we need at least to be able to get the corresponding streams dynamically.

[OpenGuidou]

Ha found it here:
io.nats.client.JetStreamManagement#getStreamNames(java.lang.String)
I'll try using that.

Thank you for the help.

[ripienaar]

@ripienaar any idea about the cli issue?

~ # nats s info
panic: runtime error: slice bounds out of range [:-2]

@scottf hard to say from that, would need to see the full backtrace and, if possible, output with --trace

[OpenGuidou]

@ripienaar any idea about the cli issue?

~ # nats s info
panic: runtime error: slice bounds out of range [:-2]

@scottf hard to say from that, would need to see the full backtrace and, if possible, output with --trace

Here is is:

~ # nats s info --trace
14:22:35 >>> $JS.API.STREAM.NAMES
{"offset":0}

14:22:35 <<< $JS.API.STREAM.NAMES
{"type":"io.nats.jetstream.api.v1.stream_names_response","total":17,"offset":0,"limit":1024,"streams":["test2-1-1","test2-1-2","test2-1-3","test2-1-4","test2-2-1","test2-2-2","test2-2-3","test2-2-4","test-error","test3-1","test3-2","test3-3","test3-4","test1-1","test1-2","test1-3","test1-4"]}

panic: runtime error: slice bounds out of range [:-2]

goroutine 1 [running]:
github.com/AlecAivazis/survey/v2.paginate(0xe57b80?, {0xc0006b2300?, 0x14?, 0x194?}, 0xc00011f200?)
        /go/pkg/mod/github.com/!alec!aivazis/survey/[email protected]/survey.go:403 +0xae
github.com/AlecAivazis/survey/v2.(*Select).Prompt(0xc00019f680, 0xc0004c69e8)
        /go/pkg/mod/github.com/!alec!aivazis/survey/[email protected]/select.go:259 +0x18c
github.com/AlecAivazis/survey/v2.Ask({0xc0004c3980, 0x1, 0x7f8d4e580dd8?}, {0xdd00c0?, 0xc0002819e0}, {0x0, 0x0, 0xc00019f680?})
        /go/pkg/mod/github.com/!alec!aivazis/survey/[email protected]/survey.go:334 +0x38c
github.com/AlecAivazis/survey/v2.AskOne(...)
        /go/pkg/mod/github.com/!alec!aivazis/survey/[email protected]/survey.go:251
github.com/nats-io/natscli/cli.askOne({0x11566d0, 0xc00019f680}, {0xdd00c0, 0xc0002819e0}, {0x0, 0x0, 0x0})
        /go/pkg/mod/github.com/nats-io/[email protected]/cli/util.go:187 +0xd6
github.com/nats-io/natscli/cli.selectStream(0x0?, {0x0, 0x0}, 0x0, 0x0)
        /go/pkg/mod/github.com/nats-io/[email protected]/cli/util.go:169 +0x1c5
github.com/nats-io/natscli/cli.(*streamCmd).connectAndAskStream(0xc0006b2000)
        /go/pkg/mod/github.com/nats-io/[email protected]/cli/stream_command.go:2625 +0xc9
github.com/nats-io/natscli/cli.(*streamCmd).infoAction(0xc0006b2000, 0x0?)
        /go/pkg/mod/github.com/nats-io/[email protected]/cli/stream_command.go:1844 +0x25
github.com/choria-io/fisk.(*actionMixin).applyActions(0xc000121248?, 0xf01220?)
        /go/pkg/mod/github.com/choria-io/[email protected]/actions.go:28 +0x5c
github.com/choria-io/fisk.(*Application).applyActions(0xc00015a640?, 0xc000132000)
        /go/pkg/mod/github.com/choria-io/[email protected]/app.go:761 +0xed
github.com/choria-io/fisk.(*Application).execute(0xc00015a640, 0xc000132000?, {0xc000089520, 0x2, 0x2})
        /go/pkg/mod/github.com/choria-io/[email protected]/app.go:599 +0x66
github.com/choria-io/fisk.(*Application).Parse(0xc00015a640, {0xc000032090?, 0x7f8d4e5775b8?, 0xd0?})
        /go/pkg/mod/github.com/choria-io/[email protected]/app.go:262 +0x14f
github.com/choria-io/fisk.(*Application).MustParseWithUsage(0xc00015a640, {0xc000032090, 0x3, 0x3})
        /go/pkg/mod/github.com/choria-io/[email protected]/app.go:827 +0x4e
main.main()
        /go/pkg/mod/github.com/nats-io/[email protected]/nats/main.go:69 +0x20ce
~ # 

[ripienaar]

Thank you @OpenGuidou that helps, I can open an issue github.com/nats-io/natscli or you can if you want, we'll get it sorted for next CLI release - though this is pretty weird must be some weird edge case in a dependency

[OpenGuidou]

Sure, you can open it. On my side it's not really blocking me as I can still to nats stream ls and nats stream info .
Thanks for checking.

[jnmoyne]

It is expected: can not have more than one stream capturing the same subject(s)

So here you have defined 3 streams capturing on "test2.*.*.*.*.1", "test2.*.*.*.*.2" and "test2.*.*.*.*.3" respectively, which means that you can not have a stream capturing "test2.>" (because that overlaps with "test2.*.*.*.*.1", "test2.*.*.*.*.2" and "test2.*.*.*.*.3") and when are doing that js.subscribe("test2.>") it looks for a stream that captures that subject and fails as it can not find one.

Instead you have to create 3 consumers (one on each partition's stream) and subscribe 3 times (you can still pass it the same callback).