JetStream and Work retention policy Stream with Durable consumers and fan-out

[riton]

JetStream and Work retention policy Stream with fan-out

Hello everyone,

I'm beginning with NATS and JetStream and I'm trying to achieve the following design:

Nothing is in production yet and I'm testing with a 3 nodes JetStream Cluster with the latest versions of NATS.

My goal in details

1. do not loose an event.
2. Any app {1..N} may go offline anytime and come back later online. I would still want them to get all the messages that were sent while they were offline.
3. The number of app is not known in advance and may grow or shrink. Adding or removing app would be a manual process so It's okay if such a thing requires human operation / administrative commands.
4. Optional: There could be multiple instances of app 1. E.g: app 1 - instance 1, app 1 - instance 2. In this case, all messages should be delivered to only one instance of app 1.

My initial design

TL;DR: I'm still learning and I've discovered that this design do not allow me to add multiple app that all see all messages.

Stream creation

I've initially created a Jetstream Stream such as:

❯ nats stream add \
  --subjects="EVENTS.*" \
  --description="All events" \
  --storage=file \
  --replicas=3 \
  --cluster=example.org \
  --ack \
  --retention=work \
  --discard=old \
  --max-msgs=-1 \
  --max-msgs-per-subject=-1 \
  --max-bytes=-1 \
  --max-age=-1 \
  --max-msg-size=-1 \
  --dupe-window=10s \
  --no-allow-rollup \
  --no-deny-delete \
  --no-deny-purge \
  EVENT_STREAM

At first glance, the Work retention policy looks like what I want.

Durable consumer creation

I've read that Consumer can be seen as a View in a Stream .

My original idea was to create multiple Consumers. One per app.

I've originally created my Consumer with:

❯ nats consumer add \
	--filter="EVENTS.*" \
	--ack=explicit \
	--deliver=all \
	--backoff=linear \
	--pull \
	--no-headers-only \
	--description="Gather all events for app 1" \
	--replay="instant" \
	--max-deliver=9999999999999 \
	--max-pending="100" \
	--max-waiting=10 \
	EVENT_STREAM CONSUMER_APP_1

Note: --max-deliver=9999999999999 was set to this ridiculously high value. If -1 of indifinetly was a valid value, I would have go for it.

So far, everything is working as expected.

My app 1 sees all messages.
If I add another instance of app 1, messages are dispatched to only one of the instance of app 1. So far perfect.

Add new app

At this moment, I was thinking that I'd finished the design, all I have to do is create a new Durable Consumer.

My original idea was to Create one "Durable Consumer" per app

So I've tried to create another Durable Consumer for app 2:

❯ nats consumer add \
	--filter="EVENTS.*" \
	--ack=explicit \
	--deliver=all \
	--backoff=linear \
	--pull \
	--no-headers-only \
	--description="Gather all events for app 2" \
	--replay="instant" \
	--max-deliver=9999999999999 \
	--max-pending="100" \
	--max-waiting=10 \
	EVENT_STREAM CONSUMER_APP_2

and I've faced the following error:

nats: error: Consumer creation failed: filtered consumer not unique on workqueue stream (10100)

My understanding of the error

At the moment, I understand that the error filtered consumer not unique on workqueue stream is raised because my Consumers are using the same --filter="EVENTS.*" and that overlapping filters are not allowed with workqueue stream.

I've read almost all documentation I can find and I'm kinda lost.

According to my understanding of the documentation, workqueue stream still looks like what I'm trying to achieve.

I understand that the subject filter in --filter="EVENTS.*" is the thing that is blocking me to create another Consumer.
I don't really see any use of subjects like EVENTS.foo or EVENTS.bar in my use case since all apps should see all events.

At the moment I'm kinda stuck and any help would be really appreciated.

Regards

Rémi

[ripienaar]

Insted of ading more consumers, connect your apps to the same consumer.

[riton]

Hi @ripienaar

Thanks for your more than very quick answer ! 🏃‍♂️

According to my tests If multiple app are connected to the same Consumer, messages will be dispatched between the apps.

I've just tested this with the following producer:

(fish shell is used here)

❯ for i in (seq 1 5); nats publish 'EVENTS.any' "[$i] $(date) / $(uuidgen)"; end
[...]
15:14:41 Published 72 bytes to "EVENTS.any"

and here are my apps.

Note: CONSUMER_ALL_APPS was just named for this test, but defined exactly as described in my original message.

app 1

❯ while true; nats consumer sub --ack EVENT_STREAM CONSUMER_ALL_APPS; sleep 2; end
[15:14:41] subj: EVENTS.any / tries: 1 / cons seq: 222 / str seq: 222 / pending: 0

[2] Thu Sep  7 15:14:41 CEST 2023 / 7dfbd0e8-4c45-4ab1-b85d-941aa5bb0e0f

Acknowledged message

[15:14:43] subj: EVENTS.any / tries: 1 / cons seq: 224 / str seq: 224 / pending: 1

[4] Thu Sep  7 15:14:41 CEST 2023 / 1aa0788f-e17c-4331-bdd8-936e0afd0654

Acknowledged message

[15:14:45] subj: EVENTS.any / tries: 1 / cons seq: 225 / str seq: 225 / pending: 0

[5] Thu Sep  7 15:14:41 CEST 2023 / 1310e4c2-bc38-4cc8-a1a5-0728e137111e

Acknowledged message

app 2

❯ while true; nats consumer sub --ack EVENT_STREAM CONSUMER_ALL_APPS; sleep 2; end
[15:14:41] subj: EVENTS.any / tries: 1 / cons seq: 221 / str seq: 221 / pending: 0

[1] Thu Sep  7 15:14:41 CEST 2023 / 10b20488-535d-41c0-be11-d149c090a087

Acknowledged message

[15:14:43] subj: EVENTS.any / tries: 1 / cons seq: 223 / str seq: 223 / pending: 2

[3] Thu Sep  7 15:14:41 CEST 2023 / 8521c868-0947-4670-a2b6-17c0d31df888

Acknowledged message

I can observe that:

• app 1 receives messages index 2, 4 and 5
• app 2 receives messages index 1 and 3

What I would like instead is:

• app 1 receives messages index 1, 2, 3, 4 and 5
• app 2 receives messages index 1, 2, 3, 4 and 5

I may have misunderstood your answer.

Regards

[ripienaar]

OK, reading again you want more than 1 app to process the same message. Then dont use WorkQueue. WorkQueue is designed for any message being processed succesfully once only.

Use Interest or limits.

With those messages will distribute within multiple instances of teh same App (they would share a consumer) and multiple apps all get all the messages (each app is a consumer)

[riton]

I'm now using limits policy and it work as expected 🥳

Thanks for your time and response 🙏