Limits-based Stream (Java)

Limits-based Stream in JetStream

To get started with JetStream, a stream must be created. The mental model for a stream is that it binds a set of subjects for which messages published to those subjects will be persisted.

A stream is implemented as an implicit server-side service that receives a request (the published message) and replies back once the message has been persisted.

There are handful of different kinds of streams and configuration options, but we will start with the most basic one having a limits-based retention policy. This policy is the default, however, limits still apply to streams with other retention policies.

The stream limit choices include:

the maximum number of messages
the maximum total size in bytes
the maximum age of a message

There is also a specialized maximum messages limit that can be applied at the subject level, but this will be demonstrated in a separate example.

By default, no limits are set which would require manually managing the ever-growing stream. However, if any of these limits satisfy how the stream should be truncated, simply turn these limits on and let the server manage everything.

In this example, we showcase the behavior or applying these limits and the flexibility of JetStream supporting dynamically changing the stream configuration on-demand.

CLI Go Python JavaScript Rust C# .NET V2 Java Ruby Elixir Crystal C

Jump to the output or the recording

$ nbe run jetstream/limits-stream/java

View the source code or learn how to run this example yourself

Code

package example;


import io.nats.client.*;
import io.nats.client.api.PublishAck;
import io.nats.client.api.StorageType;
import io.nats.client.api.StreamConfiguration;
import io.nats.client.api.StreamInfo;


import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;


public class Main {
    public static void main(String[] args) {
        String natsURL = System.getenv("NATS_URL");
        if (natsURL == null) {
            natsURL = "nats://127.0.0.1:4222";
        }

Initialize a connection to the server. The connection is AutoCloseable on exit.

        try (Connection nc = Nats.connect(natsURL)) {

Access JetStream and JetStreamManagement which provide methods to create streams and consumers as well as convenience methods for publishing to streams and consuming messages from the streams.

            JetStream js = nc.jetStream();
            JetStreamManagement jsm = nc.jetStreamManagement();

We will declare the initial stream configuration by specifying the name and subjects. Stream names are commonly uppercased to visually differentiate them from subjects, but this is not required. A stream can bind one or more subjects which almost always include wildcards. In addition, no two streams can have overlapping subjects otherwise the primary messages would be persisted twice. There are options to replicate messages in various ways, but that will be explained in later examples.

JetStream provides both file and in-memory storage options. For durability of the stream data, file storage must be chosen to survive crashes and restarts. This is the default for the stream, but we can still set it explicitly.

            String streamName = "EVENTS";
            StreamConfiguration.Builder streamConfigBuilder = StreamConfiguration.builder()
                    .name(streamName)
                    .subjects("events.>")
                    .storageType(StorageType.File);

Finally, let’s add/create the stream with the default (no) limits.

            StreamInfo stream = jsm.addStream(streamConfigBuilder.build());
            System.out.println("Created the stream.");

Let’s publish a few messages which are received by the stream since they match the subject bound to the stream. The js.publish method is a convenience for sending a nc.request and waiting for the acknowledgement.

            js.publish("events.page_loaded", null);
            js.publish("events.mouse_clicked", null);
            js.publish("events.mouse_clicked", null);
            js.publish("events.page_loaded", null);
            js.publish("events.mouse_clicked", null);
            js.publish("events.input_focused", null);
            System.out.println("Published 6 messages.");

There is also an async form in which the client can send multiple messages to the server and then asynchronously receives the acknowledgements.

            List<CompletableFuture<PublishAck>> futures = new ArrayList<>();
            futures.add(js.publishAsync("events.input_changed", null));
            futures.add(js.publishAsync("events.input_blurred", null));
            futures.add(js.publishAsync("events.key_pressed", null));
            futures.add(js.publishAsync("events.input_focused", null));
            futures.add(js.publishAsync("events.input_changed", null));
            futures.add(js.publishAsync("events.input_blurred", null));


            CompletableFuture<Void> publishCompleted = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
            try {
                publishCompleted.join();
                System.out.println("Published 6 messages.");
            } catch (Exception e) {
                System.out.println("Publish failed " + e);
            }

Checking out the stream info, we can see how many messages we have.

            printStreamState(jsm, streamName, "initial state");

Stream configuration can be dynamically changed. For example, we can set the max messages limit to 10 and it will truncate the two initial events in the stream.

Most fields can be edited, but some are not allowed to be changed. See also: https://docs.nats.io/nats-concepts/jetstream/streams#configuration.

            streamConfigBuilder.maxMessages(10);
            jsm.updateStream(streamConfigBuilder.build());
            System.out.println("Set max messages to 10.");

Checking out the info, we see there are now 10 messages and the first sequence and timestamp are based on the third message.

            printStreamState(jsm, streamName, "after setting max messages to 10");

Limits can be combined and whichever one is reached, it will be applied to truncate the stream. For example, let’s set a maximum number of bytes for the stream.

            streamConfigBuilder.maxBytes(300);
            jsm.updateStream(streamConfigBuilder.build());
            System.out.println("Set max bytes to 300.");

Inspecting the stream info we now see more messages have been truncated to ensure the size is not exceeded.

            printStreamState(jsm, streamName, "after setting max bytes to 300");

Finally, for the last primary limit, we can set the max age.

            streamConfigBuilder.maxAge(Duration.ofSeconds(1));
            jsm.updateStream(streamConfigBuilder.build());
            System.out.println("Set max age to one second.");

Looking at the stream info, we still see all the messages..

            printStreamState(jsm, streamName, "before max age removes messages");

until a second passes.

            System.out.println("Sleeping one second...");
            Thread.sleep(1000);


            printStreamState(jsm, streamName, "after max age has removed messages");
        } catch (InterruptedException | IOException | JetStreamApiException e) {
            e.printStackTrace();
        }
    }


    private static void printStreamState(JetStreamManagement jsm, String streamName, String label) throws IOException, JetStreamApiException {
        StreamInfo streamInfo = jsm.getStreamInfo(streamName);
        System.out.printf("Inspecting stream info, %s:\n%s\n\n", label, streamInfo.getStreamState());
    }
}

Output

Created the stream.
Published 6 messages.
Published 6 messages.
Inspecting stream info, initial state:
StreamState{msgs=12, bytes=594, firstSeq=1, lastSeq=12, consumerCount=0, firstTime=2024-08-08T16:15:24.030760988Z[GMT], lastTime=2024-08-08T16:15:24.035359095Z[GMT], subjectCount=6, subjects=[], deletedCount=0, deleteds=[], lostStreamData=null}

Set max messages to 10.
Inspecting stream info, after setting max messages to 10:
StreamState{msgs=10, bytes=496, firstSeq=3, lastSeq=12, consumerCount=0, firstTime=2024-08-08T16:15:24.032156468Z[GMT], lastTime=2024-08-08T16:15:24.035359095Z[GMT], subjectCount=6, subjects=[], deletedCount=0, deleteds=[], lostStreamData=null}

Set max bytes to 300.
Inspecting stream info, after setting max bytes to 300:
StreamState{msgs=6, bytes=298, firstSeq=7, lastSeq=12, consumerCount=0, firstTime=2024-08-08T16:15:24.034013677Z[GMT], lastTime=2024-08-08T16:15:24.035359095Z[GMT], subjectCount=4, subjects=[], deletedCount=0, deleteds=[], lostStreamData=null}

Set max age to one second.
Inspecting stream info, before max age removes messages:
StreamState{msgs=6, bytes=298, firstSeq=7, lastSeq=12, consumerCount=0, firstTime=2024-08-08T16:15:24.034013677Z[GMT], lastTime=2024-08-08T16:15:24.035359095Z[GMT], subjectCount=4, subjects=[], deletedCount=0, deleteds=[], lostStreamData=null}

Sleeping one second...
Inspecting stream info, after max age has removed messages:
StreamState{msgs=0, bytes=0, firstSeq=13, lastSeq=12, consumerCount=0, firstTime=1970-01-01T00:00Z[GMT], lastTime=2024-08-08T16:15:24.035359095Z[GMT], subjectCount=0, subjects=[], deletedCount=0, deleteds=[], lostStreamData=null}

Recording

Note, playback is half speed to make it a bit easier to follow.