JetStream, the persistence layer of NATS, not only allows for the higher qualities of service and features associated with 'streaming', but it also enables some functionalities not found in messaging systems.

One such feature is the Object store functionality, which allows client applications to create buckets (corresponding to streams) that can store a set of files. Files are stored and transmitted in chunks, allowing files of arbitrary size to be transferred safely over the NATS infrastructure.

Note: Object store is not a distributed storage system. All files in a bucket will need to fit on the target file system.

• Walkthrough

Basic Capabilities

The Object Store implements a chunking mechanism, allowing you to for example store and retrieve files (i.e. the object) of any size by associating them with a path or file name as the key.

• add a bucket to hold the files.

• put Add a file to the bucket

• get Retrieve the file and store it to a designated location

Advanced Capabilities

• watch Subscribe to changes in the bucket. Will receive notifications on successful put and del operations.

When a stream is configured with a source or mirror, it will automatically and asynchronously replicate messages from the origin stream.

source or mirror are designed to be robust and will recover from a loss of connection. They are suitable for geographic distribution over high latency and unreliable connections. E.g. even a leaf node starting and connecting intermittently every few days will still receive or send messages over the source/mirror link. Another use case is when connecting streams cross-account.

There are several options available when declaring the configuration.

• Name - Name of the origin stream to source messages from.

• StartSeq - An optional start sequence of the origin stream to start mirroring from.

• StartTime - An optional message start time to start mirroring from. Any messages that are equal to or greater than the start time will be included.

• FilterSubject - An optional filter subject which will include only messages that match the subject, typically including a wildcard. Note, this cannot be used with SubjectTransforms.

• SubjectTransforms - An optional set of to apply when sourcing messages from the origin stream. Note, in this context, the Source will act as a filter on the origin stream and the Destination can optionally be provided to apply a transform. Since multiple subject transforms can be used, disjoint subjects can be sourced from the origin stream while maintaining the order of the messages. Note, this cannot be used with FilterSubject.

• Domain - An optional JetStream domain of where the origin stream exists. This is commonly used in a hub cluster and leafnode topology.

The stream using a source or mirror configuration can have its own retention policy, replication, and storage type.

General behavior

• All configurations are made on the receiving side. The stream from which data is sourced and mirrored does not need to be configured. No cleanup is required on the origin side if the receiver disappears.

• A stream can be the origin (source) for multiple streams. This is useful for geographic distribution or for designing "fan out" topologies where data needs to be distributed reliable to a large number (up to millions) of client connections.

• Leaf nodes and leaf node domains are explicitly supported through the API prefix

Source specific

A stream defining Sources is a generalized replication mechanism and allows for sourcing data from one or more streams concurrently. A stream with sources can still act as a regular stream allowing direct write/publish by local clients to the stream. Essentially the source streams and local client writes are aggregated into a single interleaved stream. Combined with subject transformation and filtering sourcing allows to design sophisticated data distribution architectures.

Mirror specific

A mirror can source its messages from exactly one stream and a clients can not directly write to the mirror. Although messages cannot be published to a mirror directly by clients, messages can be deleted on-demand (beyond the retention policy), and consumers have all capabilities available on regular streams.

Expected behavior in edge conditions

• Source and mirror contracts are designed with one-way (geographic) data replication in mind. Neither configuration provides a full synchronization between streams, which would include deletes or replication of other stream attributes.

• The content of the stream from which a source or mirror is drawn needs to be reasonable stable. Quickly deleting messages after publishing them may result in inconsistent replication due to the asynchronous nature of the replication process.

• Sources and Mirror try to be be efficient in replicating messages and are lenient towards the source/mirror origin being unreachable (event for extended periods of time), e.g. when using leaf nodes, which are connected intermittently. For sake of efficiency the recovery interval in case of a disconnect is 10-20s.

WorkQueue retention

Source and mirror works with origin stream with workqueue retention in a limited context. The source/mirror will act as a consumer removing messages from the origin stream.

The implementation is not resilient when connecting over intermittent leaf node connections though. Within a cluster where the target stream (with the source/mirror agreement) it will generally work well.

Interest base retention

NATS has a built-in persistence engine called JetStream which enables messages to be stored and replayed at a later time. Unlike NATS Core which requires you to have an active subscription to process messages as they happen, JetStream allows the NATS server to capture messages and replay them to consumers as needed. This functionality enables a different quality of service for your NATS messages, and enables fault-tolerant and high-availability configurations.

JetStream is built into nats-server. If you have a cluster of JetStream-enabled servers you can enable data replication and thus guard against failures and service disruptions.

JetStream was created to address the problems identified with streaming technology today - complexity, fragility, and a lack of scalability. Some technologies address these better than others, but no current streaming technology is truly multi-tenant, horizontally scalable, or supports multiple deployment models. No other technology that we are aware of can scale from edge to cloud using the same security context while having complete deployment observability for operations.

Additional capabilities enabled by JetStream

The JetStream persistence layer enables additional use cases typically not found in messaging systems. Being built on top of JetStream they inherit the core capabilities of JetStream, replication, security, routing limits, and mirroring.

• A map (associative array) with atomic operations

• File transfer, replications and storage API. Uses chunked transfers for scalability.

Key/Value and File transfer are capabilities commonly found in in-memory databases or deployment tools. While NATS does not intend to compete with the feature set of such tools, it is our goal to provide the developer with reasonable complete set of data storage and replications features for use cases like micro service, edge deployments and server management.

Configuration

To configure a nats-server with JetStream refer to:

Examples

For runnable JetStream code examples, refer to .

Goals

JetStream was developed with the following goals in mind:

• The system must be easy to configure and operate and be observable.

• The system must be secure and operate well with NATS 2.0 security models.

• The system must scale horizontally and be applicable to a high ingestion rate.

JetStream capabilities

Streaming: temporal decoupling between the publishers and subscribers

One of the tenets of basic publish/subscribe messaging is that there is a required temporal coupling between the publishers and the subscribers: subscribers only receive the messages that are published when they are actively connected to the messaging system (i.e. they do not receive messages that are published while they are not subscribing or not running or disconnected). The traditional way for messaging systems to provide temporal decoupling of the publishers and subscribers is through the 'durable subscriber' functionality or sometimes through 'queues', but neither one is perfect:

• durable subscribers need to be created before the messages get published

• queues are meant for workload distribution and consumption, not to be used as a mechanism for message replay.

However, in many use cases, you do not need to 'consume exactly once' functionality but rather the ability to replay messages on demand, as many times as you want. This need has led to the popularity of some 'streaming' messaging platforms.

JetStream provides both the ability to consume messages as they are published (i.e. 'queueing') as well as the ability to replay messages on demand (i.e. 'streaming'). See below.

Replay policies

JetStream consumers support multiple replay policies, depending on whether the consuming application wants to receive either:

• all of the messages currently stored in the stream, meaning a complete 'replay' and you can select the 'replay policy' (i.e. the speed of the replay) to be either:

  • instant (meaning the messages are delivered to the consumer as fast as it can take them).

  • original (meaning the messages are delivered to the consumer at the rate they were published into the stream, which can be very useful for example for staging production traffic).

Retention policies and limits

JetStream enables new functionalities and higher qualities of service on top of the base 'Core NATS' functionality. However, practically speaking, streams can't always just keep growing 'forever' and therefore JetStream supports multiple retention policies as well as the ability to impose size limits on streams.

Limits

You can impose the following limits on a stream

• Maximum message age.

• Maximum total stream size (in bytes).

• Maximum number of messages in the stream.

You must also select a discard policy which specifies what should happen once the stream has reached one of its limits and a new message is published:

• discard old means that the stream will automatically delete the oldest message in the stream to make room for the new messages.

• discard new means that the new message is discarded (and the JetStream publish call returns an error indicating that a limit was reached).

Retention policy

You can choose what kind of retention you want for each stream:

• limits (the default) is to provide a replay of messages in the stream.

• work queue (the stream is used as a shared queue and messages are removed from it as they are consumed) is to provide the exactly-once consumption of messages in the stream.

• interest (messages are kept in the stream for as long as there are consumers that haven't delivered the message yet) is a variation of work queue that only retains messages if there is interest (consumers currently defined on the stream) for the message's subject.

Note that regardless of the retention policy selected, the limits (and the discard policy) always apply.

Subject mapping transformations

JetStream also enables the ability to apply subject mapping transformations to messages as they are ingested into a stream.

Persistent and Consistent distributed storage

You can choose the durability as well as the resilience of the message storage according to your needs.

• Memory storage.

• File storage.

• Replication (1 (none), 2, 3) between nats servers for Fault Tolerance.

JetStream uses a NATS optimized RAFT distributed quorum algorithm to distribute the persistence service between NATS servers in a cluster while maintaining immediate consistency (as opposed to ) even in the face of failures.

For writes (publications to a stream), the formal consistency model of NATS JetStream is . On the read side (listening to or replaying messages from streams) the formal models don't really apply because JetStream does not support atomic batching of multiple operations together (so the only kind of 'transaction' is the persisting, replicating and voting of a single operation on the stream) but in essence, JetStream is because messages are added to a stream in one global order (which you can control using compare and publish).

Do note, while we do guarantee immediate consistency when it comes to and . We don't guarantee at this time, as reads through direct get requests may be served by followers or mirrors. More consistent results can be achieved by sending get requests to the stream leader.

JetStream can also provide encryption at rest of the messages being stored.

In JetStream the configuration for storing messages is defined separately from how they are consumed. Storage is defined in a and consuming messages is defined by multiple .

Stream replication factor

A stream's replication factor (R, often referred to as the number 'Replicas') determines how many places it is stored allowing you to tune to balance risk with resource usage and performance. A stream that is easily rebuilt or temporary might be memory-based with a R=1 and a stream that can tolerate some downtime might be file-based R-1.

Typical usage to operate in typical outages and balance performance would be a file-based stream with R=3. A highly resilient, but less performant and more expensive configuration is R=5, the replication factor limit.

Rather than defaulting to the maximum, we suggest selecting the best option based on the use case behind the stream. This optimizes resource usage to create a more resilient system at scale.

• Replicas=1 - Cannot operate during an outage of the server servicing the stream. Highly performant.

• Replicas=2 - No significant benefit at this time. We recommend using Replicas=3 instead.

• Replicas=3 - Can tolerate the loss of one server servicing the stream. An ideal balance between risk and performance.

Mirroring and Sourcing between streams

JetStream also allows server administrators to easily mirror streams, for example between different JetStream domains in order to offer disaster recovery. You can also define a stream that 'sources' from one or more other streams.

Syncing data to disk

JetStream’s file-based streams persist messages to disk. However, while JetStream does flush file writes to the OS synchronously, under the default configuration it does not immediately fsync data to disk. The server uses a configurable sync_interval option, with a default value of 2 minutes, which controls how often the server will fsync its data. The data will be fsync-ed no later than this interval. This has important consequences for durability with respect to OS failures (meaning ungraceful exit of the Operating System such as a power outage, and not just ungraceful exit or killing of the nats-server process itself):

In a non-replicated setup, an OS failure may result in data loss. A client might publish a message and receive an acknowledgment, but the data may not yet be safely stored to disk. As a result, after an OS failure recovery, a server may have lost recently acknowledged messages.

In a replicated setup, a published message is acknowledged after it successfully replicated to at least a quorum of servers. However, replication alone is not enough to guarantee the strongest level of durability against multiple systemic failures.

• If multiple servers fail simultaneously, all due to an OS failure, and before their data has been fsync-ed, the cluster may fail to recover the most recently acknowledged messages.

• If a failed server lost data locally due to an OS failure, although extremely rare, there are some combinations of events where it may rejoin the cluster and form a new majority with nodes that have never received or persisted a given message. The cluster may then proceed with incomplete data causing acknowledged messages to be lost.

Setting a lower sync_interval increases the frequency of disk writes, and reduces the window for potential data loss, but at the expense of performance. Additionally, setting sync_interval: always will make sure servers fsync after every message before it is acknowledged. This setting, combined with replication in different data centers or availability zones, provides the strongest durability guarantees but at the slowest performance.

The default settings have been chosen to balance performance and risk of data loss in what we consider to be a typical production deployment scenario across multiple availability zones.

For example, consider a stream with 3 replicas deployed across three separate availability zones. For the stream state to diverge across nodes would require that:

• One of the 3 servers is already offline, isolated or partitioned.

• A second server’s OS needs to fail such that it loses writes of messages that were only available on 2 out of 3 nodes due to them not being fsync-ed.

• The stream leader that’s part of the above 2 out of 3 nodes needs to go down or become isolated/partitioned.

In the end, 2 out of 3 nodes will be available, the previous stream leader with the writes will be unavailable, one server will have lost some writes due to the OS failure, and one server will have never seen these writes due to the earlier partition. The last two servers could then form a majority and accept new writes, essentially losing some of the former writes.

Importantly this is a failure condition where stream state could diverge, but in a system that is deployed across multiple availability zones, it would require multiple faults to align precisely in the right way.

A potential mitigation to a failure of this kind is not automatically bringing back a server process that was OS-failed until it is known that a majority of the remaining servers have received the new writes, or by peer-removing the crashed server and admitting it as a new and wiped peer and allowing it to recover over the network from existing healthy nodes (although this could be expensive depending on the amount of data involved).

For use cases where minimizing loss is an absolute priority, sync_interval: always can of course still be configured, but note that this will have a server-wide performance impact that may affect throughput or latencies. For production environments, operators should evaluate whether the default is correct for their use case, target environment, costs, and performance requirements.

Alternatively, a hybrid approach can be used where existing clusters still function under their default sync_interval settings but a new cluster gets added that’s configured with sync_interval: always, and utilizes server tags. The placement of a stream can then be specified to have this stream store data on this higher durability cluster through the use of .

Create a replicated stream that’s specifically placed in the cluster using sync_interval: always, to ensure the strongest durability only for stream writes that require this level of durability.

De-coupled flow control

JetStream provides decoupled flow control over streams, the flow control is not 'end to end' where the publisher(s) are limited to publish no faster than the slowest of all the consumers (i.e. the lowest common denominator) can receive but is instead happening individually between each client application (publishers or consumers) and the nats server.

When using the JetStream publish calls to publish to streams there is an acknowledgment mechanism between the publisher and the NATS server, and you have the choice of making synchronous or asynchronous (i.e. 'batched') JetStream publish calls.

On the subscriber side, the sending of messages from the NATS server to the client applications receiving or consuming messages from streams is also flow controlled.

Exactly once semantics

Because publications to streams using the JetStream publish calls are acknowledged by the server the base quality of service offered by streams is 'at least once', meaning that while reliable and normally duplicate free there are some specific failure scenarios that could result in a publishing application believing (wrongly) that a message was not published successfully and therefore publishing it again, and there are failure scenarios that could result in a client application's consumption acknowledgment getting lost and therefore in the message being re-sent to the consumer by the server. Those failure scenarios while being rare and even difficult to reproduce do exist and can result in perceived 'message duplication' at the application level.

Therefore, JetStream also offers an 'exactly once' quality of service. For the publishing side, it relies on the publishing application attaching a unique message or publication ID in a message header and on the server keeping track of those IDs for a configurable rolling period of time in order to detect the publisher publishing the same message twice. For the subscribers a double acknowledgment mechanism is used to avoid a message being erroneously re-sent to a subscriber by the server after some kinds of failures.

Consumers

JetStream are 'views' on a stream, they are subscribed to (or pulled) by client applications to receive copies of (or to consume if the stream is set as a working queue) messages stored in the stream.

Fast push consumers

Client applications can choose to use fast un-acknowledged push (ordered) consumers to receive messages as fast as possible (for the selected replay policy) on a specified delivery subject or to an inbox. Those consumers are meant to be used to 'replay' rather than 'consume' the messages in a stream.

Horizontally scalable pull consumers with batching

Client applications can also use and share pull consumers that are demand-driven, support batching and must explicitly acknowledge message reception and processing which means that they can be used to consume (i.e. use the stream as a distributed queue) as well as process the messages in a stream.

Pull consumers can and are meant to be shared between applications (just like queue groups) in order to provide easy and transparent horizontal scalability of the processing or consumption of messages in a stream without having (for example) to worry about having to define partitions or worry about fault-tolerance.

Note: using pull consumers doesn't mean that you can't get updates (new messages published into the stream) 'pushed' in real-time to your application, as you can pass a (reasonable) timeout to the consumer's Fetch call and call it in a loop.

Consumer acknowledgments

While you can decide to use un-acknowledged consumers trading quality of service for the fastest possible delivery of messages, most processing is not idem-potent and requires higher qualities of service (such as the ability to automatically recover from various failure scenarios that could result in some messages not being processed or being processed more than once) and you will want to use acknowledged consumers. JetStream supports more than one kind of acknowledgment:

• Some consumers support acknowledging all the messages up to the sequence number of the message being acknowledged, some consumers provide the highest quality of service but require acknowledging the reception and processing of each message explicitly as well as the maximum amount of time the server will wait for an acknowledgment for a specific message before re-delivering it (to another process attached to the consumer).

• You can also send back negative acknowledgements.

• You can even send

Key Value Store

The JetStream persistence layer enables the Key Value store: the ability to store, retrieve and delete value messages associated with a key into a bucket.

Watch and History

You can subscribe to changes in a Key Value on the bucket or individual key level with watch and optionally retrieve a history of the values (and deletions) that have happened on a particular key.

Atomic updates and locking

The Key Value store supports atomic create and update operations. This enables pessimistic locks (by creating a key and holding on to it) and optimistic locks (using CAS - compare and set).

Object Store

The Object Store is similar to the Key Value Store. The key being replaced by a file name and value being designed to store arbitrarily large objects (e.g. files, even if they are very large) rather than 'values' that are message-sized (i.e. limited to 1Mb by default). This is achieved by chunking messages.

Legacy

Note that JetStream completely replaces the legacy NATS streaming layer.

Consider this architecture

While it is an incomplete architecture it does show a number of key points:

• Many related subjects are stored in a Stream

• Consumers can have different modes of operation and receive just subsets of the messages

The following is a small walkthrough on creating a stream and a consumer and interacting with the stream using the nats cli.

Prerequisite: enabling JetStream

If you are running a local nats-server stop it and restart it with JetStream enabled using nats-server -js (if that's not already done)

You can then check that JetStream is enabled by using

If you see the below then JetStream is not enabled

1. Creating a stream

Let's start by creating a stream to capture and store the messages published on the subject "foo".

Enter nats stream add <Stream name> (in the examples below we will name the stream "my_stream"), then enter "foo" as the subject name and hit return to use the defaults for all the other stream attributes:

You can then check the information about the stream you just created:

2. Publish some messages into the stream

Let's now start a publisher

As messages are being published on the subject "foo" they are also captured and stored in the stream, you can check that by using nats stream info my_stream and even look at the messages themselves using nats stream view my_stream or nats stream get my_stream

3. Creating a consumer

Now at this point if you create a 'Core NATS' (i.e. non-streaming) subscriber to listen for messages on the subject 'foo', you will only receive the messages being published after the subscriber was started, this is normal and expected for the basic 'Core NATS' messaging. In order to receive a 'replay' of all the messages contained in the stream (including those that were published in the past) we will now create a 'consumer'

We can administratively create a consumer using the 'nats consumer add ' command, in this example we will name the consumer "pull_consumer", and we will leave the delivery subject to 'nothing' (i.e. just hit return at the prompt) because we are creating a 'pull consumer' and select all for the start policy, you can then just use the defaults and hit return for all the other prompts. The stream the consumer is created on should be the stream 'my_stream' we just created above.

You can check on the status of any consumer at any time using nats consumer info or view the messages in the stream using nats stream view my_stream or nats stream get my_stream, or even remove individual messages from the stream using nats stream rmm

3. Subscribing from the consumer

Now that the consumer has been created and since there are messages in the stream we can now start subscribing to the consumer:

This will print out all the messages in the stream starting with the first message (which was published in the past) and continuing with new messages as they are published until the count is reached.

Note that in this example we are creating a pull consumer with a 'durable' name, this means that the consumer can be shared between as many consuming processes as you want. For example instead of running a single nats consumer next with a count of 1000 messages you could have started two instances of nats consumer each with a message count of 500 and you would see the consumption of the messages from the consumer distributed between those instances of nats

Replaying the messages again

Once you have iterated over all the messages in the stream with the consumer, you can get them again by simply creating a new consumer or by deleting that consumer (nats consumer rm) and re-creating it (nats consumer add).

4. Cleaning up

You can clean up a stream (and release the resources associated with it (e.g. the messages stored in the stream)) using nats stream purge

You can also delete a stream (which will also automatically delete all of the consumers that may be defined on that stream) using nats stream rm

The Key/Value Store is a JetStream feature, so we need to verify it is enabled by

which may return

In this case, you should enable JetStream.

Prerequisite: enabling JetStream

If you are running a local nats-server stop it and restart it with JetStream enabled using nats-server -js (if that's not already done)

You can then check that JetStream is enabled by using

Creating a KV bucket

A 'KV bucket' is like a stream; you need to create it before using it, as in nats kv add <KV Bucket Name>:

Storing a value

Now that we have a bucket, we can assign, or 'put', a value to a specific key:

which should return the key's value Value1

Getting a value

We can fetch, or 'get', the value for a key "Key1":

Deleting a value

You can always delete a key and its value by using

It is harmless to delete a non-existent key (check this!!).

Atomic operations

K/V Stores can also be used in concurrent design patterns, such as semaphores, by using atomic 'create' and 'update' operations.

E.g. a client wanting exclusive use of a file can lock it by creating a key, whose value is the file name, with create and deleting this key after completing use of that file. A client can increase the reslience against failure by using a timeout for the bucket containing this key. The client can use update with a revision number to keep the bucket alive.

Updates can also be used for more fine-grained concurrency control, sometimes known as optimistic locking, where multiple clients can try a task, but only one can successfully complete it.

Create (aka exclusive locking)

Create a lock/semaphore with the create operation.

Only one create can succeed. First come, first serve. All concurrent attempts will result in an error until the key is deleted

Update with CAS (aka optimistic locking)

We can also atomically update, sometimes known as a CAS (compare and swap) operation, a key with an additional parameter revision

A second attempt with the same revision 13, will fail

Watching a K/V Store

An unusual functionality of a K/V Store is being able to 'watch' a bucket, or a specific key in that bucket, and receive real-time updates to changes in the store.

For the example above, run nats kv watch my-kv. This will start a watcher on the bucket we have just created earlier. By default, the KV bucket has a history size of one, and so it only remembers the last change. In our case, the watcher should see a delete of the value associated with the key "Key1":

If we now concurrently change the value of 'my-kv' by

The watcher will see that change:

Cleaning up

When you are finished using a bucket, you can delete the bucket, and its resources, by using the rm operator:

JetStream, the persistence layer of NATS, not only allows for the higher qualities of service and features associated with 'streaming', but it also enables some functionalities not found in messaging systems.

One such feature is the Key/Value store functionality, which allows client applications to create buckets and use them as immediately (as opposed to eventually) consistent, persistent associative arrays (or maps). Note that this is an abstraction on top of the Stream functionality. Buckets are materialized as Streams (with a name starting with KV_), everything you can do with a bucket you can do with a Stream, but you ultimately have more functionality and flexibility and control when using the Stream functionality directly.

Do note, while we do guarantee immediate consistency when it comes to monotonic writes and monotonic reads. We don't guarantee read your writes at this time, as reads through direct get requests may be served by followers or mirrors. More consistent results can be achieved by sending get requests to the underlying stream leader of the Key/Value store.

Managing a Key Value store

1. Create a bucket, which corresponds to a stream in the underlying storage. Define KV/Stream limits as appropriate

2. Use the operation below.

Map style operations

You can use KV buckets to perform the typical operations you would expect from an immediately consistent key/value store:

• put: associate a value with a key

• get: retrieve the value associated with a key

• delete: clear any value associated with a key

Atomic operations used for locking and concurrency control

• create: associate the value with a key only if there is currently no value associated with that key (i.e. compare to null and set)

• update: compare and set (aka compare and swap) the value for a key

Limiting size, TTL etc.

You can set limits for your buckets, such as:

• the maximum size of the bucket

• the maximum size for any single value

• a TTL: how long the store will keep values for

Treating the Key Value store as a message stream

Finally, you can even do things that typically can not be done with a Key/Value Store:

• watch: watch for changes happening for a key, which is similar to subscribing (in the publish/subscribe sense) to the key: the watcher receives updates due to put or delete operations on the key pushed to it in real-time as they happen

• watch all: watch for all the changes happening on all the keys in the bucket

• history: retrieve a history of the values (and delete operations) associated with each key over time (by default the history of buckets is set to 1, meaning that only the latest value/operation is stored)

Notes

A valid key can contain the following characters: a-z, A-Z, 0-9, _, -, ., = and /, i.e. it can be a dot-separated list of tokens (which means that you can then use wildcards to match hierarchies of keys when watching a bucket). The value can be any byte array.

Streams with source and mirror configurations are best managed through a client API. If you intend to create such a configuration from command line with NATS CLI you should use a JSON configuration.

Example stream configuration with two sources

Minimal example

With additional options

If you are running a local nats-server stop it and restart it with JetStream enabled using nats-server -js (if that's not already done)

You can then check that JetStream is enabled by using

Which should output something like:

If you see the below instead then JetStream is not enabled

A consumer is a stateful view of a stream. It acts as an interface for clients to consume a subset of messages stored in a stream and will keep track of which messages were delivered and acknowledged by clients.

Unlike Core NATS, which provides an at most once delivery guarantee, a consumer in JetStream can provide an at least once delivery guarantee.

While Streams are responsible for storing the published messages, the consumer is responsible for tracking the delivery and acknowledgments. This tracking ensures that if a message is not acknowledged (un-acked or 'nacked'), the consumer will automatically attempt to re-deliver it. JetStream consumers support various acknowledgment types and policies. If a message is not acknowledged within a user-specified number of delivery attempts, an advisory notification is emitted.

Dispatch type - Pull / Push

Consumers can be push-based where messages will be delivered to a specified subject or pull-based which allows clients to request batches of messages on demand. The choice of what kind of consumer to use depends on the use-case.

If there is a need to process messages in an application controlled manner and easily scale horizontally, you would use a 'pull consumer'. A simple client application that wants a replay of messages from a stream sequentially you would use an 'ordered push consumer'. An application that wants to benefit from load balancing or acknowledge messages individually will use a regular push consumer.

Ordered Consumers

Ordered consumers are the convenient default type of push & pull consumers designed for applications that want to efficiently consume a stream for data inspection or analysis.

• Always ephemeral

• No acknowledgements (if gap is detected, consumer is recreated)

• Automatic flow control/pull processing

Persistence - Durable / Ephemeral

In addition to the choice of being push or pull, a consumer can also be ephemeral or durable. A consumer is considered durable when an explicit name is set on the Durable field when creating the consumer, or when InactiveThreshold is set.

Durables and ephemeral have the same message delivery semantics but an ephemeral consumer will not have persisted state or fault tolerance (server memory only) and will be automatically cleaned up (deleted) after a period of inactivity, when no subscriptions are bound to the consumer.

By default, consumers will have the same replication factor as the stream they consume, and will remain even when there are periods of inactivity (unless InactiveThreshold is set explicitly). Consumers can recover from server and client failure.

Configuration

Below are the set of consumer configuration options that can be defined. The Version column indicates the version of nats-server in which the option was introduced. The Editable column indicates the option can be edited after the consumer is created.

General

AckPolicy

The policy choices include:

• AckExplicit: The default policy. Each individual message must be acknowledged. Recommended for most reliability and functionality.

• AckNone: No acknowledgment needed; the server assumes acknowledgment on delivery.

• AckAll: Acknowledge only the last message received in a series; all previous messages are automatically acknowledged. Will acknowledge all pending messages for all subscribers for Pull Consumer.

If an acknowledgment is required but not received within the AckWait window, the message will be redelivered.

Warning: The server may consider an acknowledgment arriving out of the window. For instance, in a queue situation, if a first process fails to acknowledge within the window and the message has been redelivered to another consumer, the acknowledgment from the first consumer will be considered.

DeliverPolicy

The policy choices include:

• DeliverAll: Default policy. Start receiving from the earliest available message in the stream.

• DeliverLast: Start with the last message added to the stream, or the last message matching the consumer's filter subject if defined.

• DeliverLastPerSubject

MaxAckPending

The MaxAckPending capability provides flow control and applies to both push and pull consumers. For push consumers, MaxAckPending is the only form of flow control. For pull consumers, client-driven message delivery creates implicit one-to-one flow control with subscribers.

For high throughput, set MaxAckPending to a high value. For applications with high latency due to external services, use a lower value and adjust AckWait to avoid re-deliveries.

FilterSubjects

A filter subject provides server-side filtering of messages before delivery to clients.

For example, a stream factory-events with subject factory-events.*.* can have a consumer factory-A with a filter factory-events.A.* to deliver only events for factory A.

A consumer can have a singular FilterSubject or plural FilterSubjects. Multiple filters can be applied, such as [factory-events.A.*, factory-events.B.*] or specific event types [factory-events.*.item_produced, factory-events.*.item_packaged].

Warning: For granular consumer permissions, a single filter uses $JS.API.CONSUMER.CREATE.{stream}.{consumer}.{filter} to restrict users to specific filters. Multiple filters use the general $JS.API.CONSUMER.DURABLE.CREATE.{stream}.{consumer}, which does not include the {filter} token. Use a different strategy for granular permissions.

Pull-specific

These options apply only to pull consumers. For configuration examples, see .

Push-specific

These options apply only to push consumers. For configuration examples, see .

Streams are message stores, each stream defines how messages are stored and what the limits (duration, size, interest) of the retention are. Streams consume normal NATS subjects, any message published on those subjects will be captured in the defined storage system. You can do a normal publish to the subject for unacknowledged delivery, though it’s better to use the JetStream publish calls instead as the JetStream server will reply with an acknowledgement that it was successfully stored.

The diagram above shows the concept of storing all ORDERS.* in the Stream even though there are many types of order related messages. We’ll show how you can selectively consume subsets of messages later. Relatively speaking the Stream is the most resource consuming component so being able to combine related data in this manner is important to consider.

Streams can consume many subjects. Here we have ORDERS.* but we could also consume SHIPPING.state into the same Stream should that make sense.

Stream limits and message retention

Streams support various retention policies which define when messages in the stream can be automatically deleted, such as when stream limits are hit (like max count, size or age of messages), or also more novel options that apply on top of the limits such as interest-based retention or work-queue semantics (see ).

Upon reaching message limits, the server will automatically discard messages either by removing the oldest messages to make room for new ones (DiscardOld) or by refusing to store new messages (DiscardNew). For more details, see .

Streams support deduplication using a Nats-Msg-Id header and a sliding window within which to track duplicate messages. See the section.

For examples on how to configure streams with your preferred NATS client, see .

Configuration

Below are the set of stream configuration options that can be defined. The Version column indicates the version of the server the option was introduced. The Editable column indicates the option can be edited after the stream created. See client-specific examples .

StorageType

The storage types include:

• File (default) - Uses file-based storage for stream data.

• Memory - Uses memory-based storage for stream data.

Subjects

Note: a stream configured as a cannot be configured with a set of subjects. A mirror implicitly sources a subset of the origin stream (optionally with a filter), but does not subscribe to additional subjects.

If no explicit subject is specified, the default subject will be the same name as the stream. Multiple subjects can be specified and edited over time. Note, if messages are stored by a stream on a subject that is subsequently removed from the stream config, consumers will still observe those messages if their subject filter overlaps.

RetentionPolicy

The retention options include:

• LimitsPolicy (default) - Retention based on the various limits that are set including: MaxMsgs, MaxBytes, MaxAge, and MaxMsgsPerSubject. If any of these limits are set, whichever limit is hit first will cause the automatic deletion of the respective message(s). See a .

• WorkQueuePolicy - Retention with the typical behavior of a FIFO queue. Each message can be consumed only once. This is enforced by only allowing

DiscardPolicy

The discard behavior applies only for streams that have at least one limit defined. The options include:

• DiscardOld (default) - This policy will delete the oldest messages in order to maintain the limit. For example, if MaxAge is set to one minute, the server will automatically delete messages older than one minute with this policy.

• DiscardNew - This policy will reject new messages from being appended to the stream if it would exceed one of the limits. An extension to this policy is DiscardNewPerSubject which will apply this policy on a per-subject basis within the stream.

Placement

Refers to the placement of the stream assets (data) within a NATS deployment, be it a single cluster or a supercluster. A given stream, including all replicas (not mirrors), are bound to a single cluster. So when creating or moving a stream, a cluster will be chosen to host the assets.

Without declaring explicit placement for a stream, by default, the stream will be created within the cluster that the client is connected to assuming it has sufficient storage available.

By declaring stream placement, where these assets are located can be controlled explicitly. This is generally useful to co-locate with the most active clients (publishers or consumers) or may be required for data sovereignty reasons.

Placement is supported in all client SDKs as well as the CLI. For example, adding a stream via the CLI to place a stream in a specific cluster looks like this:

For this to work, all servers in a given cluster must define the name field within the server configuration block.

If you have multiple clusters that form a supercluster, then each is required to have a different name.

Another placement option are tags. Each server can have its own set of tags, , typically describing properties of geography, hosting provider, sizing tiers, etc. In addition, tags are often used in conjunction with the jetstream.unique_tag config option to ensure that replicas must be placed on servers having different values for the tag.

For example, a server A, B, and C in the above cluster might all the same configuration except for the availability zone they are deployed to.

Now we can create a stream by using tags, for example indicating we want a stream in us-east1.

If we had a second cluster in Google Cloud with the same region tag, the stream could be placed in either the AWS or GCP cluster. However, the unique_tag constraint ensures each replica will be placed in a different AZ in the cluster that was selected implicitly by the placement tags.

Although less common, note that both the cluster and tags can be used for placement. This would be used if a single cluster contains servers have different properties.

Sources and Mirrors

When a stream is configured with a source or mirror, it will automatically and asynchronously replicate messages from the origin stream. There are several options when declaring the configuration.

A source or mirror stream can have its own retention policy, replication, and storage type. Changes to the source or mirror, e.g. deleting messages or publishing, do not reflect on the origin stream.

Stream sources

A stream defining Sources is a generalized replication mechanism and allows for sourcing data from one or more streams concurrently as well as allowing direct write/publish by clients. Essentially the source streams and client writes are aggregated into a single interleaved stream. Subject transformation and filtering allow for powerful data distribution architectures.

Mirrors

A mirror can source its messages from exactly one stream and a clients can not directly write to the mirror. Although messages cannot be published to a mirror directly by clients, messages can be deleted on-demand (beyond the retention policy), and consumers have all capabilities available on regular streams.

For details see:

AllowRollup

If enabled, the AllowRollup stream option allows for a published message having a Nats-Rollup header indicating all prior messages should be purged. The scope of the purge is defined by the header value, either all or sub.

The Nats-Rollup: all header will purge all prior messages in the stream. Whereas the sub value will purge all prior messages for a given subject.

A common use case for rollup is for state snapshots, where the message being published has accumulated all the necessary state from the prior messages, relative to the stream or a particular subject.

RePublish

If enabled, the RePublish stream option will result in the server re-publishing messages received into a stream automatically and immediately after a successful write, to a distinct destination subject.

For high scale needs where, currently, a dedicated consumer may add too much overhead, clients can establish a core NATS subscription to the destination subject and receive messages that were appended to the stream in real-time.

The fields for configuring republish include:

• Source - An optional subject pattern which is a subset of the subjects bound to the stream. It defaults to all messages in the stream, e.g. >.

• Destination - The destination subject messages will be re-published to. The source and destination must be a valid .

For each message that is republished, a set of are automatically added.

Note: The reply-subject is being removed on republished messages, even if no-ack is set on the stream.

SubjectTransform

If configured, the SubjectTransform will perform a subject transform to matching subjects of messages received by the stream and transform the subject, before storing it in the stream. The transform configuration specifies a Source and Destination field, following the rules of .

Message headers are used in a variety of JetStream contexts, such de-duplication, auto-purging of messages, metadata from republished messages, and more.

Nats- is a reserved namespace. Please use a different prefix for your own headers. This list may not be complete. Additional headers may be used for API internal messages or messages used for monitoring and control.

Publish

Headers that can be set by a client when a message being published. These headers are recognized by the server.

RePublish or direct get

When messages are being re-published (must be configured in stream settings) from a stream or retrieved with a direct get operation from stream these headers are being set.

Do not set these headers on client published messages.

Sources

Headers that are implicitly added to messages sourced from other streams.

The format of the header content may change in the future. Please parse conservatively and assume that additional fields may be added or that older nats-server version have fewer fields.

Tracing

When tracing is activated every subsystem that touches a message will produce Trace Events. These Events are aggregated per server and published to a destination subject.

Note that two variants exist. traceparent as per the trace context standard and ad hoc tracing through Nats-Trace-Dest.

Introduced in version 2.11 - see

Scheduler

Message scheduling in streams. Needs to be enabled on the stream with "allow schedules" flag.

Introduced in version 2.11 - see

The final scheduled message will contain the following headers.

Batch send

Introduced in version 2.12 with optimizations in 2.14 - see

Atomic batch sends will use the following headers. Batches are atomic on send only, but a client may reconstruct a batch using the headers below.

Internal

Headers used internally by API clients and the server. Should not be set by user.

This is list is not exhaustive. Headers used in error replies may not be documented.

Nats-Counter-Sources

Mirror

Headers used for internal flow-control messages for a mirror.

This is for information only and may change without notice.