NATS Jetstream Internals

[beatrausch]

Hi all,

thx for building nats, it’s a great piece of software.

I would like to learn a bit more about the internals of nats, especially the design of the Jetstream. What I understood so far based on the docs: There are multiple raft groups that are responsible for replicating/storing certain aspects of the Jetstream like streams, consumer and meta-data.

Can you share some more details about the Jetstream design? Especially, how data is stored? Is the data of raft groups stored in a partitioned manner? I am basically familiar with raft based databased systems. Typically, there are tables, tables are split into tablets, tables are replicated based on raft. How does it work in nats Jetstream?

According to the documentation the recommend number of Jetstream enabled servers is 3/5. Does it mean this is recommend maximum number of Jetstream enabled servers? If so, I am not sure how this affects scalability, because this would mean, that all related Jetstream data hast to be stored on the Jetstream servers.

Regards

[ripienaar]

Regarding the limits - 3 or 5 are recommended levels per stream (5 is the enforced max). That means the stream will inhabit 5 servers but it could be in a much larger cluster over all. With streams spread out over more but never exceeding the limit per stream.

Otherwise as you say each stream has a raft group and so does most types of consumer (ephemeral are not raft based).

The meta cluster handles placement of streams and some other things like stats and various API calls and all servers (even more than 5) belong to it.

[beatrausch]

Regarding the limits - 3 or 5 are recommended levels per stream (5 is the enforced max). That means the stream will inhabit 5 servers but it could be in a much larger cluster over all. With streams spread out over more but never exceeding the limit per stream.

With limits you mean the replication factor per stream or is there some kind of distribution policy (how many nodes are involved in storing a single stream)?

What is still unclear to me, if there is a kind of data partitioning within a stream. If not, it would mean, that the size of a single stream is limited by the size of the persistence of a single node?

[ripienaar]

Yes a single stream is limited to the size of a node.

You can use aomething like subject mappings to automatically partition incoming data into different streams spread theough the cluster

[beatrausch]

Hi,

after reading the documentation I came up with the following sample, that should be transparent for subscribing applications, as long as ordering per partition/stream is no hard requirement:

func main() 
{
  natsURL := os.Getenv("NATS_URL")
  if natsURL == "" {
    natsURL = nats.DefaultURL
  }

  nc, _ := nats.Connect(natsURL)
	js, _ := nc.JetStream()


  // based on subject mappings: 
  //  "events.*.*" : "events.{{partition(2,1)}}.{{wildcard(1)}}.{{wildcard(2)}}"
  // create 2 partitions
  
	// Create a stream for each region.
	js.AddStream(&nats.StreamConfig{
		Name:     "EVENTS-0",
		Subjects: []string{"events.0.>"},
	})

	js.AddStream(&nats.StreamConfig{
		Name:     "EVENTS-1",
		Subjects: []string{"events.1.>"},
	})

	// Create a consumer for each stream. Both publish to the same deliver subject.
	js.AddConsumer("EVENTS-0", &nats.ConsumerConfig{
		Durable:        "event-processor",
		DeliverSubject: "partioned-events",
		DeliverGroup:   "event-processor",
		AckPolicy:      nats.AckExplicitPolicy,
	})

	js.AddConsumer("EVENTS-1", &nats.ConsumerConfig{
		Durable:        "event-processor",
		DeliverSubject: "partioned-events",
		DeliverGroup:   "event-processor",
		AckPolicy:      nats.AckExplicitPolicy,
	})

	// Publish messages to each stream.
	js.Publish("events.eu.page_loaded", nil)
	js.Publish("events.eu.input_focused", nil)
	js.Publish("events.us.page_loaded", nil)
	js.Publish("events.us.mouse_clicked", nil)
	js.Publish("events.eu.mouse_clicked", nil)
	js.Publish("events.us.input_focused", nil)

	// Subscribe to the deliver subject with core NATS subscription. Observe that
	// messages from both streams are being received and can be ack'ed.
	sub, _ := js.QueueSubscribeSync("partioned-events", "event-processor")
	defer sub.Drain()

	for {
		msg, err := sub.NextMsg(time.Second)
		if err == nats.ErrTimeout {
			break
		}

		fmt.Println(msg.Subject)
		msg.Ack()
	}
}

What do you think?

[beatrausch]

@ripienaar
How are writes (publish) and reads (subscription) are handled in terms of raft? I guess publish calls are always coordinated via the raft leader and reads are served by the followers as well?

[ripienaar]

Write completes when every replica thats available committed it. Reads are spread around the replicas. You can see in a consumer report who is the leader for a consumer, here C1 consumer is served by a follower.

[beatrausch]

Thanks for the feedback, it helps a lot to get a better understanding how nats jetstream works.

@ripienaar Before I mark the question as answered, cloud you please provide a short feedback, on the sample code regarding stream portioning.

[ripienaar]

Code seems fine, you can combine all the partitions into one stream - later spread to multiple if they are huge or you want placement flexibility for each partition. Anyway yes in general you got the idea.