Quarkus Reactive Messaging Nats Jetstream

This extension allow usage of NATS JetStream inside a Quarkus App, in JVM and Native mode.

The extension implements a new connector type quarkus-jetstream in SmallRye Reactive Messaging that will use the NATS client.

The client and configuration were significantly reworked after version 3.38.0. If you are using 3.38.0 or earlier, see the 3.38.0 documentation instead of this page. Upgrading from 3.38.0? Follow the Migration Guide.

Installation

If you want to use this extension, you need to add the io.quarkiverse.reactivemessaging.nats-jetstream:quarkus-reactive-messaging-nats-jetstream extension first to your build file.

For instance, with Maven, add the following dependency to your POM file:

<dependency>
    <groupId>io.quarkiverse.reactivemessaging.nats-jetstream</groupId>
    <artifactId>quarkus-messaging-nats-jetstream</artifactId>
    <version>3.39.3</version>
</dependency>

Then configure your application by adding the NATS JetStream connector type:

# Inbound
mp.messaging.incoming.[channel-name].connector=quarkus-jetstream

# Outbound
mp.messaging.outgoing.[channel-name].connector=quarkus-jetstream

Receiving messages from NATS JetStream

Let’s imagine you have a NATS JetStream broker running, and accessible using the localhost:4242 address. Configure your application to receive NATS messages on the data channel from the stream named: test and the subject named: data as follows:

quarkus.messaging.nats.connection.servers=nats://localhost:4242
quarkus.messaging.nats.connection.username=guest
quarkus.messaging.nats.connection.password=guest
quarkus.messaging.nats.connection.ssl-enabled=false

# The stream and its consumer are created if they don't already exist on NATS
quarkus.messaging.nats.streams.test.name=test
quarkus.messaging.nats.streams.test.subjects=data

quarkus.messaging.nats.consumers.data-consumer.stream=test
quarkus.messaging.nats.consumers.data-consumer.name=data-consumer
quarkus.messaging.nats.consumers.data-consumer.filter-subject=data
quarkus.messaging.nats.consumers.data-consumer.durable=true

mp.messaging.incoming.data.connector=quarkus-jetstream
mp.messaging.incoming.data.stream=test
mp.messaging.incoming.data.consumer=data-consumer

Then, your application receives Message<Data>. You can consumes the payload directly:

package inbound;

import org.eclipse.microprofile.reactive.messaging.Incoming;

import jakarta.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class DataConsumer {

    @Incoming("data")
    public void consume(Data data) {
        // process your data.
    }

}

If you want more examples, please take a look at the tests of the extension.

Request/Reply (RequestReply)

Instead of plain pub/sub, you can use request/reply semantics: inject RequestReply<Req, Rep> on an outgoing channel and call request(). Each call publishes the payload on the channel’s subject and returns a Uni that completes when the matching reply arrives, or fails with TimeoutException if no reply arrives within reply.timeout.

The stream MUST cover both the request subject and the reply subject. If the reply subject is not covered by the stream, replies can never be persisted or delivered and every request fails with a timeout (NATS happily accepts creating the consumer either way). If the stream does not exist at all, the first request fails with ConsumerManagementException; the next request re-subscribes automatically.

Requestor

Configure an outgoing channel for the requests:

mp.messaging.outgoing.orders.connector=quarkus-jetstream
mp.messaging.outgoing.orders.stream=orders
mp.messaging.outgoing.orders.subject=orders
# Optional - defaults to '<subject>.replies'
mp.messaging.outgoing.orders.reply.subject=orders.replies
# Optional - default 5000 ms
mp.messaging.outgoing.orders.reply.timeout=10000

Then inject the requestor and send requests:

package outbound;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;

import org.eclipse.microprofile.reactive.messaging.Channel;

import io.quarkiverse.reactive.messaging.nats.jetstream.connector.reply.RequestReply;
import io.smallrye.mutiny.Uni;

@ApplicationScoped
public class OrderService {

    @Inject
    @Channel("orders")
    RequestReply<String, String> requestor;

    public Uni<String> placeOrder(String order) {
        return requestor.request(order);
    }

}

Replier

The replier is a plain @Incoming/@Outgoing method. The connector reads the reply subject advertised by the requestor (the message.reply-subject header) and routes the returned payload there automatically, echoing the correlation id (message.correlation-id) so concurrent requestors never receive each other’s replies:

package inbound;

import jakarta.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

@ApplicationScoped
public class OrderReplier {

    @Incoming("orders-in")
    @Outgoing("orders-out")
    public String reply(String order) {
        return "accepted:" + order;
    }

}

Bind the channels to the same stream:

mp.messaging.incoming.orders-in.connector=quarkus-jetstream
mp.messaging.incoming.orders-in.stream=orders
mp.messaging.incoming.orders-in.consumer=orders-replier

mp.messaging.outgoing.orders-out.connector=quarkus-jetstream
mp.messaging.outgoing.orders-out.stream=orders
mp.messaging.outgoing.orders-out.subject=orders

Requestor channel configuration

Property Name Description Type Default Value

Property Name	Description	Type	Default Value
mp.messaging.outgoing.[channel-name].reply.subject	The subject on which replies are expected	String	`<subject>.replies`
mp.messaging.outgoing.[channel-name].reply.timeout	How long to wait for a reply in milliseconds	Long	5000
mp.messaging.outgoing.[channel-name].reply.inactive-threshold	How long NATS keeps an idle request/reply consumer before reclaiming it, in milliseconds	Long	60000
mp.messaging.outgoing.[channel-name].reply.correlation-id.handler	The `@Identifier` of the `CorrelationIdHandler` bean used to generate and parse correlation ids	String	uuid
mp.messaging.outgoing.[channel-name].reply.failure.handler	The `@Identifier` of the `ReplyFailureHandler` bean used to turn reply payloads into failures	String

mp.messaging.outgoing.[channel-name].reply.subject

The subject on which replies are expected

String

<subject>.replies

mp.messaging.outgoing.[channel-name].reply.timeout

How long to wait for a reply in milliseconds

Long

5000

mp.messaging.outgoing.[channel-name].reply.inactive-threshold

How long NATS keeps an idle request/reply consumer before reclaiming it, in milliseconds

Long

60000

mp.messaging.outgoing.[channel-name].reply.correlation-id.handler

The @Identifier of the CorrelationIdHandler bean used to generate and parse correlation ids

String

uuid

mp.messaging.outgoing.[channel-name].reply.failure.handler

The @Identifier of the ReplyFailureHandler bean used to turn reply payloads into failures

String

The correlation id travels in the NATS message header message.correlation-id; the requestor advertises its reply subject via the message.reply-subject header. Both names are fixed for this version.

Scaling model and failure semantics

Every requestor instance creates its own non-durable consumer on the shared reply subject (deliver_policy=new, max_deliver=1) and receives every reply; replies are demultiplexed by correlation id and non-matches are acknowledged and discarded. This trades a little extra network traffic for zero instance identity: requestors and repliers can scale independently.
Idle consumers are reclaimed by NATS after reply.inactive-threshold, so no consumer leaks accumulate.
If publishing the request fails, the returned Uni fails with PublishException.
If a reply payload should be treated as an error, provide a CDI bean implementing ReplyFailureHandler (selected per channel via reply.failure.handler) and return the failure from it; the caller’s Uni fails with that error.
To observe in-flight requests, use RequestReply.getPendingReplies(); pending entries are removed on completion, failure, timeout and shutdown.

Using NATS Key/Value store

If you want to use the Key/Value store in NATS, then configure your application as follows:

quarkus.messaging.nats.connection.servers=nats://localhost:4242
quarkus.messaging.nats.connection.username=guest
quarkus.messaging.nats.connection.password=guest
quarkus.messaging.nats.connection.ssl-enabled=false

# Creates a "test" bucket stored in memory. This part is optional. If you already have created a bucket on NATS
# then this configuration is redundant.
quarkus.messaging.nats.key-values.test.bucket-name=test
quarkus.messaging.nats.key-values.test.storage-type=Memory
quarkus.messaging.nats.key-values.test.description=Test bucket

Then your application can use the Key/Value store like this, through Client.keyValue(bucketName):

@Path("/key-value")
@RequestScoped
class KeyValueStoreResource {
    private final Client client;

    @Inject
    public KeyValueStoreResource(Client client) {
        this.client = client;
    }

    @GET
    @Path("{key}")
    public Uni<byte[]> getValue(@PathParam("key") String key) {
        return client.keyValue("test").get(key).onItem().transform(entry -> entry.value().orElse(null));
    }

    @PUT
    @Path("{key}")
    @Consumes("application/octet-stream")
    public Uni<Void> putValue(@PathParam("key") String key, byte[] value) {
        return client.keyValue("test").put(key, value).replaceWithVoid();
    }

    @DELETE
    @Path("{key}")
    public Uni<Void> deleteValue(@PathParam("key") String key) {
        return client.keyValue("test").delete(key);
    }
}

Using NATS Object Store

If you want to use the Object store in NATS, then configure your application as follows:

quarkus.messaging.nats.connection.servers=nats://localhost:4242
quarkus.messaging.nats.connection.username=guest
quarkus.messaging.nats.connection.password=guest
quarkus.messaging.nats.connection.ssl-enabled=false

# Creates a "test" object store. This part is optional. If you already have created the bucket on NATS
# then this configuration is redundant.
quarkus.messaging.nats.object-stores.test.bucket-name=test
quarkus.messaging.nats.object-stores.test.description=Test object store

Then your application can use the Object store like this, through Client.objectStore(bucketName):

@Path("/object-store")
@RequestScoped
class ObjectStoreResource {
    private final Client client;

    @Inject
    public ObjectStoreResource(Client client) {
        this.client = client;
    }

    @GET
    @Path("{name}")
    public Uni<byte[]> getObject(@PathParam("name") String name) {
        return client.objectStore("test").get(name).onItem().transform(ObjectEntry::data);
    }

    @PUT
    @Path("{name}")
    @Consumes("application/octet-stream")
    public Uni<Void> putObject(@PathParam("name") String name, byte[] data) {
        return client.objectStore("test").put(name, data).replaceWithVoid();
    }

    @DELETE
    @Path("{name}")
    public Uni<Void> deleteObject(@PathParam("name") String name) {
        return client.objectStore("test").delete(name).replaceWithVoid();
    }
}

Serializing Java Objects

By default, Java objects are serialized using java.io.* streams. If either the quarkus-jackson or quarkus-jsonb dependency is present, objects are serialized as JSON instead.

To use a custom serializer, set the build-time property quarkus.messaging.nats.serializer to the fully-qualified class name of a class that implements io.quarkiverse.reactive.messaging.nats.jetstream.client.message.Serializer. The default value is io.quarkiverse.reactive.messaging.nats.jetstream.client.message.JacksonSerializer.

quarkus.messaging.nats.serializer=com.example.MySerializer

Open Telemetry Tracing

To use Open Telemetry tracing, add the quarkus-opentelemetry dependency.

Configuration

The stream, consumer, key-value store and object store configurations are all optional. Each declared entry is created on NATS if it does not already exist (addIfAbsent); if the configuration for a given name is omitted entirely, it is assumed to already exist on NATS.

Table 1. NATS connection configuration
Property Name	Description	Type	Default Value
quarkus.messaging.nats.connection.servers	A comma-separated list of URI’s nats://{host}:{port} to use for establishing the initial connection to the NATS cluster.	String	nats://localhost:4222
quarkus.messaging.nats.connection.username	The username to connect to the NATS server	String
quarkus.messaging.nats.connection.password	The password to connect to the NATS server	String
quarkus.messaging.nats.connection.token	The token to connect to the NATS server	String
quarkus.messaging.nats.connection.credential-path	The path to the credentials file to connect to the NATS server	String
quarkus.messaging.nats.connection.ssl-enabled	Whether to enable SSL/TLS secure connections to the NATS server	Boolean	false
quarkus.messaging.nats.connection.connection-timeout	The connection timeout	Duration
quarkus.messaging.nats.connection.error-listener	The classname for the error listener	String
quarkus.messaging.nats.connection.buffer-size	The size in bytes to make buffers for connections	Integer
quarkus.messaging.nats.connection.tls-algorithm	The TLS algorithm	String
quarkus.messaging.nats.connection.connection-attempts	The maximum number of attempts to attempt to re-connect to NATS	Integer	-1 (unlimited)
quarkus.messaging.nats.connection.connection-backoff	Back-off delay between to attempt to re-connect to NATS	Duration	1s
quarkus.messaging.nats.connection.tls-configuration-name	The name of the TLS configuration (bucket) used for client authentication in the TLS registry	String
quarkus.messaging.nats.connection.inbox-prefix	Retrieves the optional inbox prefix used for communication. The inbox prefix allows customization of the subscriber’s inbox name.	String

Each entry in the quarkus.messaging.nats.streams map is created (if absent) with [stream-name] as its map key; name must be set explicitly and is conventionally the same value as [stream-name].

Table 2. NATS JetStream Stream configuration
Property Name	Description	Type	Default Value
quarkus.messaging.nats.streams.[stream-name].name	The name of the stream	String
quarkus.messaging.nats.streams.[stream-name].retention-policy	Declares the retention policy for the stream	Limits, Interest or WorkQueue	Interest
quarkus.messaging.nats.streams.[stream-name].compression	The compression used for stream data	None or S2	None
quarkus.messaging.nats.streams.[stream-name].storage-type	The storage type for stream data	File or Memory	File
quarkus.messaging.nats.streams.[stream-name].discard-policy	The discard policy for this stream	New or Old	Old
quarkus.messaging.nats.streams.[stream-name].description	Description of the stream	String
quarkus.messaging.nats.streams.[stream-name].max-consumers	The maximum number of consumers for this stream	Long
quarkus.messaging.nats.streams.[stream-name].max-messages	The maximum number of messages for this stream	Long
quarkus.messaging.nats.streams.[stream-name].max-messages-per-subject	The maximum number of messages per subject for this stream	Long
quarkus.messaging.nats.streams.[stream-name].max-bytes	The maximum number of bytes for this stream	Long
quarkus.messaging.nats.streams.[stream-name].max-age	The maximum message age for this stream	Duration
quarkus.messaging.nats.streams.[stream-name].maximum-message-size	The maximum message size for this stream	Integer
quarkus.messaging.nats.streams.[stream-name].replicas	The number of replicas a message must be stored on	Integer	1
quarkus.messaging.nats.streams.[stream-name].no-ack	Whether acknowledgements are disabled for this stream	Boolean	true
quarkus.messaging.nats.streams.[stream-name].template-owner	The template json for this stream	String
quarkus.messaging.nats.streams.[stream-name].duplicate-window	The duplicate checking window for the stream. `PT0S` (zero) means duplicate checking is not enabled	Duration
quarkus.messaging.nats.streams.[stream-name].subjects	A comma separated list of the subject names bound to the stream	String
quarkus.messaging.nats.streams.[stream-name].placement	Placement directives to consider when placing replicas of this stream, as a nested group (`.cluster`, `.tags`); random placement when unset	Group
quarkus.messaging.nats.streams.[stream-name].republish	Republish configuration, as a nested group (`.source`, `.destination`, `.headers-only`)	Group
quarkus.messaging.nats.streams.[stream-name].subject-transform	Subject transform applied to the stream, as a nested group (`.source`, `.destination`)	Group
quarkus.messaging.nats.streams.[stream-name].consumer-limits	Default limits applied to consumers of this stream, as a nested group (`.max-ack-pending`, `.inactive-threshold`)	Group
quarkus.messaging.nats.streams.[stream-name].mirror	Configures this stream to mirror another stream, as a nested group (`.source-name`, `.name`, `.start-sequence`, `.start-time`, `.filter-subject`, `.external.`, `.subject-transforms`, `.consumer-source.`)	Group
quarkus.messaging.nats.streams.[stream-name].sources.[n]	Sources this stream from one or more other streams; each entry has the same shape as `mirror`	List of Group
quarkus.messaging.nats.streams.[stream-name].sealed	Whether the stream is sealed (no further changes to messages allowed)	Boolean	false
quarkus.messaging.nats.streams.[stream-name].allow-rollup	Whether the stream allows rollup	Boolean	false
quarkus.messaging.nats.streams.[stream-name].allow-direct	Whether the stream allows direct message access	Boolean	false
quarkus.messaging.nats.streams.[stream-name].mirror-direct	Whether the stream allows higher performance and unified direct access for mirrors as well	Boolean	false
quarkus.messaging.nats.streams.[stream-name].deny-delete	Whether message delete is denied for the stream	Boolean	false
quarkus.messaging.nats.streams.[stream-name].deny-purge	Whether purge is denied for the stream	Boolean	false
quarkus.messaging.nats.streams.[stream-name].discard-new-per-subject	Whether the discard-new policy with max messages per subject is applied per subject	Boolean	false
quarkus.messaging.nats.streams.[stream-name].first-sequence	The first sequence used in the stream	Long	1
quarkus.messaging.nats.streams.[stream-name].subject-delete-marker-ttl	The subject delete marker TTL	Duration
quarkus.messaging.nats.streams.[stream-name].allow-message-ttl	Whether per-message TTL is allowed	Boolean	false
quarkus.messaging.nats.streams.[stream-name].allow-message-schedules	Whether message schedules are allowed	Boolean	false
quarkus.messaging.nats.streams.[stream-name].allow-message-counter	Whether the stream is a counter stream	Boolean	false
quarkus.messaging.nats.streams.[stream-name].allow-atomic-publish	Whether atomic (batch) publish is allowed	Boolean	false
quarkus.messaging.nats.streams.[stream-name].allow-batched	Whether batched publish is allowed	Boolean	false
quarkus.messaging.nats.streams.[stream-name].persist-mode	The persist mode used by the server when storing messages	Default or Async	Default
quarkus.messaging.nats.streams.[stream-name].metadata.[key]	Application-defined metadata for the stream	Map<String, String>

Consumer configuration

Unlike streams, consumers are configured as a single, flat map under quarkus.messaging.nats.consumers, each entry naming the stream it is bound to; a channel then references a pre-configured consumer by name via mp.messaging.incoming.[channel-name].consumer. Consumers are pull-based when only pull-options (or nothing) is set; push-options is used by consumers created outside of a regular incoming channel (for example the internal request/reply reply consumer).

Table 3. NATS JetStream Consumer configuration
Property Name	Description	Type	Default Value
quarkus.messaging.nats.consumers.[consumer-name].stream	The name of the stream this consumer is bound to	String
quarkus.messaging.nats.consumers.[consumer-name].name	The name of the consumer	String
quarkus.messaging.nats.consumers.[consumer-name].durable	Set to true if the consumer should be durable	Boolean	false
quarkus.messaging.nats.consumers.[consumer-name].description	Description of the consumer	String
quarkus.messaging.nats.consumers.[consumer-name].filter-subject	A subject that overlaps with the subjects bound to the stream to filter delivery to subscribers. Cannot be used together with `filter-subjects`	String
quarkus.messaging.nats.consumers.[consumer-name].filter-subjects	A comma separated list of subjects that overlap with the subjects bound to the stream to filter delivery to subscribers. Cannot be used together with `filter-subject`	String
quarkus.messaging.nats.consumers.[consumer-name].deliver-policy	The point in the stream to receive messages from	All, Last, New, ByStartSequence, ByStartTime or LastPerSubject	All
quarkus.messaging.nats.consumers.[consumer-name].start-sequence	Used with the `ByStartSequence` deliver policy	Long	0
quarkus.messaging.nats.consumers.[consumer-name].start-time	Used with the `ByStartTime` deliver policy	ZonedDateTime
quarkus.messaging.nats.consumers.[consumer-name].acknowledge-wait	The duration the server will wait for an ack for any individual message once it has been delivered to a consumer. If an ack is not received in time, the message will be redelivered	Duration
quarkus.messaging.nats.consumers.[consumer-name].inactive-threshold	Duration that instructs the server to clean up the consumer once inactive for that long	Duration
quarkus.messaging.nats.consumers.[consumer-name].max-acknowledge-pending	The maximum number of messages, without an acknowledgement, that can be outstanding	Long
quarkus.messaging.nats.consumers.[consumer-name].max-deliver	The maximum number of times a specific message delivery will be attempted	Long
quarkus.messaging.nats.consumers.[consumer-name].backoff	A comma separated list of durations controlling the re-delivery of messages on acknowledgement timeout	List of Duration
quarkus.messaging.nats.consumers.[consumer-name].replay-policy	Whether messages are replayed at the original publish rate (`Original`) or as fast as possible (`Instant`)	Instant or Original	Instant
quarkus.messaging.nats.consumers.[consumer-name].replicas	The number of replicas for the consumer’s state. Defaults to inheriting the number of replicas from the stream	Integer
quarkus.messaging.nats.consumers.[consumer-name].memory-storage	If set, forces the consumer state to be kept in memory rather than inherit the storage type of the stream	Boolean	false
quarkus.messaging.nats.consumers.[consumer-name].sample-frequency	The percentage of acknowledgements to sample for observability, e.g. `30` or `30%`	String
quarkus.messaging.nats.consumers.[consumer-name].headers-only	Delivers only the headers of messages, not the bodies	Boolean	false
quarkus.messaging.nats.consumers.[consumer-name].pause-until	The time until the consumer is paused	ZonedDateTime
quarkus.messaging.nats.consumers.[consumer-name].metadata.[key]	Application-defined metadata for the consumer	Map<String, String>
quarkus.messaging.nats.consumers.[consumer-name].pull-options.max-waiting	The number of pulls that can be outstanding on this pull consumer	Long
quarkus.messaging.nats.consumers.[consumer-name].pull-options.max-expires	The max expire time the server allows on pull requests	Duration
quarkus.messaging.nats.consumers.[consumer-name].pull-options.max-batch	The maximum batch size a single pull request can make	Long
quarkus.messaging.nats.consumers.[consumer-name].pull-options.max-bytes	The maximum total bytes that can be requested in a given batch	Long
quarkus.messaging.nats.consumers.[consumer-name].push-options.deliver-subject	The subject messages are pushed to	String
quarkus.messaging.nats.consumers.[consumer-name].push-options.deliver-group	The optional deliver (queue) group to join	String
quarkus.messaging.nats.consumers.[consumer-name].push-options.flow-control	Enables per-subscription flow control using a sliding-window protocol	Boolean	false
quarkus.messaging.nats.consumers.[consumer-name].push-options.idle-heartbeat	If set, the server regularly sends a status message while there are no new messages to send, so the client knows the service is still up	Duration
quarkus.messaging.nats.consumers.[consumer-name].push-options.rate-limit	Used to throttle the delivery of messages to the consumer, in bits per second	Long

At the channel level, mp.messaging.incoming.[channel-name] always binds to the consumer by name as a pull consumer (see the "Channel configuration" section below); push-options only applies to consumers created directly through the Client API, such as the request/reply reply consumer.

Channel configuration

Table 4. Subscriber processor attributes (outgoing channels)
Property Name	Description	Type	Default Value
mp.messaging.outgoing.[channel-name].stream	The stream to publish messages to	String
mp.messaging.outgoing.[channel-name].subject	The subject to publish messages to	String
mp.messaging.outgoing.[channel-name].retry-backoff	The retry backoff in milliseconds for retrying failed publishes	Long	10000
mp.messaging.outgoing.[channel-name].datasource	The name of the datasource (as configured under `quarkus.messaging.nats.data-sources`) to use	String

Table 5. Publisher processor attributes (incoming channels)
Property Name	Description	Type	Default Value
mp.messaging.incoming.[channel-name].stream	The stream to consume messages from	String
mp.messaging.incoming.[channel-name].consumer	The name of the pre-configured consumer to bind to (see the "Consumer configuration" section above)	String
mp.messaging.incoming.[channel-name].batch-size	The number of messages pulled per fetch request	Integer	100
mp.messaging.incoming.[channel-name].timeout	The timeout in milliseconds for pulling messages	Long	1000
mp.messaging.incoming.[channel-name].retry-backoff	The retry backoff in milliseconds for retrying the subscription after a failure	Long	10000
mp.messaging.incoming.[channel-name].payload-type	The class name of the payload type	String
mp.messaging.incoming.[channel-name].datasource	The name of the datasource (as configured under `quarkus.messaging.nats.data-sources`) to use	String

Key/Value Store configuration

Each entry in the quarkus.messaging.nats.key-values map is created (if absent) with [bucket-name] as its map key; bucket-name must be set explicitly and is conventionally the same value as [bucket-name].

Table 6. NATS Key/Value Store configuration
Property Name	Description	Type	Default Value
quarkus.messaging.nats.key-values.[bucket-name].bucket-name	The name of the bucket	String
quarkus.messaging.nats.key-values.[bucket-name].description	Description of the Key/Value store	String
quarkus.messaging.nats.key-values.[bucket-name].storage-type	The storage type	File or Memory	File
quarkus.messaging.nats.key-values.[bucket-name].max-bucket-size	The maximum number of bytes for this bucket	Long
quarkus.messaging.nats.key-values.[bucket-name].maximum-value-size	The maximum size for an individual value in the bucket	Integer
quarkus.messaging.nats.key-values.[bucket-name].max-history-per-key	The maximum amount of history for any one key, from 1 to 64 inclusive. Includes the current value	Integer	64
quarkus.messaging.nats.key-values.[bucket-name].ttl	The maximum age for a value in this bucket	Duration
quarkus.messaging.nats.key-values.[bucket-name].limit-marker-ttl	The limit marker TTL	Duration
quarkus.messaging.nats.key-values.[bucket-name].replicas	The number of replicas for this bucket	Integer	1
quarkus.messaging.nats.key-values.[bucket-name].compression	Whether to use compression	Boolean	false
quarkus.messaging.nats.key-values.[bucket-name].republish	Republish configuration, as a nested group (`.source`, `.destination`, `.headers-only`)	Group
quarkus.messaging.nats.key-values.[bucket-name].placement	Placement directives to consider when placing replicas of this bucket, as a nested group (`.cluster`, `.tags`); random placement when unset	Group
quarkus.messaging.nats.key-values.[bucket-name].mirror	Configures this bucket to mirror another bucket/stream; same shape as the stream’s `mirror`	Group
quarkus.messaging.nats.key-values.[bucket-name].sources.[n]	Sources this bucket from one or more other buckets/streams; same shape as the stream’s `sources`	List of Group
quarkus.messaging.nats.key-values.[bucket-name].metadata.[key]	Application-defined metadata for the bucket	Map<String, String>

Object Store configuration

Each entry in the quarkus.messaging.nats.object-stores map is created (if absent) with [bucket-name] as its map key; bucket-name must be set explicitly and is conventionally the same value as [bucket-name].

Table 7. NATS Object Store configuration
Property Name	Description	Type	Default Value
quarkus.messaging.nats.object-stores.[bucket-name].bucket-name	The name of the bucket	String
quarkus.messaging.nats.object-stores.[bucket-name].description	Description of the object store	String
quarkus.messaging.nats.object-stores.[bucket-name].storage-type	The storage type	File or Memory	File
quarkus.messaging.nats.object-stores.[bucket-name].max-bucket-size	The maximum number of bytes for this bucket	Long
quarkus.messaging.nats.object-stores.[bucket-name].ttl	The maximum age for a value in this bucket	Duration
quarkus.messaging.nats.object-stores.[bucket-name].replicas	The number of replicas for this bucket	Integer	1
quarkus.messaging.nats.object-stores.[bucket-name].compression	Whether to use compression	Boolean	false
quarkus.messaging.nats.object-stores.[bucket-name].placement	Placement directives to consider when placing replicas of this bucket, as a nested group (`.cluster`, `.tags`); random placement when unset	Group
quarkus.messaging.nats.object-stores.[bucket-name].metadata.[key]	Application-defined metadata for the bucket	Map<String, String>

Dev service configuration

Property Name	Description	Type	Default Value
quarkus.messaging.nats.devservices.port	Fixed port the dev service will listen to	Integer
quarkus.messaging.nats.devservices.image-name	The image to use	String	nats:2.11
quarkus.messaging.nats.devservices.shared	Indicates if the NATS JetStream broker managed by Quarkus Dev Services is shared	Boolean	true
quarkus.messaging.nats.devservices.service-name	This property is used when you need multiple shared NATS JetStream brokers.	String	nats
quarkus.messaging.nats.devservices.enabled	If Dev Services for NATS JetStream has been explicitly enabled or disabled	Boolean	true
quarkus.messaging.nats.devservices.tls-configuration.certificate-file	The absolute path to the PEM certificate file	String
quarkus.messaging.nats.devservices.tls-configuration.key-file	The absolute path to the PEM key file	String

Property Name

Description

Type

Default Value

quarkus.messaging.nats.devservices.port

Fixed port the dev service will listen to

Integer

quarkus.messaging.nats.devservices.image-name

The image to use

String

nats:2.11

quarkus.messaging.nats.devservices.shared

Indicates if the NATS JetStream broker managed by Quarkus Dev Services is shared

Boolean

true

quarkus.messaging.nats.devservices.service-name

This property is used when you need multiple shared NATS JetStream brokers.

String

nats

quarkus.messaging.nats.devservices.enabled

If Dev Services for NATS JetStream has been explicitly enabled or disabled

Boolean

true

quarkus.messaging.nats.devservices.tls-configuration.certificate-file

The absolute path to the PEM certificate file

String

quarkus.messaging.nats.devservices.tls-configuration.key-file

The absolute path to the PEM key file

String

NATS JetStream

This extension uses the NATS JetStream client to connect to a NATS JetStream broker.

Further documentation can be found at:

Reactive Messaging

This extension uses SmallRye Reactive Messaging to build data streaming applications.

If you want to go further, check the documentation of SmallRye Reactive Messaging, the implementation used in Quarkus.