This section shows how to start a broker instance for local development(the broker Docker container and the client application are assumed to run on thesame host).

The following command creates a one-time Docker container to run RabbitMQ:

The previous command exposes only the stream port (5552), you can exposeports for other protocols:

Refer to the officialRabbitMQ Docker image web pageto find out more about its usage.

Once the container is started,the stream plugin must be enabled:

This is the simplest way to run the broker locally.The container uses thehost network,this is perfect for experimenting locally.

Once the container is started,the stream plugin must be enabled:

The container will use the following ports: 5552 (for stream) and 5672 (for AMQP.)

<dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>stream-client</artifactId><version>1.3.0</version></dependency></dependencies>

dependencies {  compile"com.rabbitmq:stream-client:1.3.0"}

<repositories><repository><id>central-portal-snapshots</id><url>https://central.sonatype.com/repository/maven-snapshots/</url><snapshots><enabled>true</enabled></snapshots><releases><enabled>false</enabled></releases></repository></repositories>

repositories {  maven {    name ='Central Portal Snapshots'    url ='https://central.sonatype.com/repository/maven-snapshots/'// Only search this repository for the specific dependency    content {      includeModule("com.rabbitmq","stream-client")    }  }  mavenCentral()}

The environment serves as the main entry point to a node or a cluster of nodes.Producer andConsumer instances are created from anEnvironment instance.Here is the simplest way to create anEnvironment instance:

Always close the environment to release resources when finished.

Treat the environment as a long-lived object.An application will usually create oneEnvironment instance when it starts up and close it when it exits.

It is possible to use a URI to specify all the necessary information toconnect to a node:

The previous snippet uses a URI that specifies the following information: host, port,username, password, and virtual host (/, which is encoded as%2f).The URI follows the same rules as theAMQP 0.9.1 URI,except the protocol must berabbitmq-stream.TLS is enabled by using therabbitmq-stream+tls scheme in the URI.

When using one URI, the corresponding node will be the main entry point to connect to.TheEnvironment will then use the stream protocol to find out more about streams topology (leaders and replicas) when asked to createProducer andConsumer instances.

If this node fails, theEnvironment will lose connectivity.To improve resilience, specify multiple URIs as fallback options:

By specifying several URIs, the environment will try to connect to the first one, andwill pick a new URI randomly in case of disconnection.

Creating the environment to connect to a cluster node usually works seamlessly.Creating publishers and consumers may encounter connection issues because the client relies on cluster hints to locate stream leaders and replicas.

These connection hints can be accurate or less appropriate depending on the infrastructure.If you encounter connection problems (such as unresolvable hostnames), thisblog post explains the root causes and solutions.Setting theadvertised_host andadvertised_portconfiguration entries should solve the most common connection problems.

To make the local development experience simple, the client library can choose to always uselocalhost for producers and consumers.This happens if the following conditions are met: the initial host to connect to islocalhost, the user isguest, and no custom address resolver has been provided.Provide a pass-throughAddressResolver toEnvironmentBuilder#addressResolver(AddressResolver) to avoid this behavior.It is unlikely this behavior applies for any real-world deployment, wherelocalhost and/or the defaultguest user should not be used.

TLS can be enabled by using therabbitmq-stream+tls scheme in the URI.The default TLS port is 5551.

Use theEnvironmentBuilder#tls method to configure TLS.The most important setting is aio.netty.handler.ssl.SslContext instance, which is created and configured with theio.netty.handler.ssl.SslContext#forClient method.Note hostname verification is enabled by default.

The following snippet shows a common configuration, whereby the client is instructed to trust servers with certificates signed by the configured certificate authority (CA).

Checking the identity of the server the client connects to is an important part of the TLS handshake.To make this work with the stream client library, it is critical the configured trusted certificates match the hosts returned by cluster nodes in the connection hints.Make sure to read the section onconnection logic.You may have to configure theadvertised_hostbroker setting in case of a mismatch between trusted certificates and the default connection hints cluster nodes return.

It is sometimes handy to trust any server certificates in development environments.EnvironmentBuilder#tls provides thetrustEverything method to do so.This should not be used in a production environment.

The following table sums up the main settings to create anEnvironment:

A load balancer can misguide the client when it tries to connect to nodes that host stream leaders and replicas.The"Connecting to Streams" blog post covers why client applications must connect to the appropriate nodes in a cluster and how aload balancer can make things complicated for them.

TheEnvironmentBuilder#addressResolver(AddressResolver) method allows intercepting the node resolution after metadata hints and before connection.Applications can use this hook to ignore metadata hints and always use the load balancer, as illustrated in the following snippet:

Note the example above sets the number of locator connections the environment maintains.Locator connections are used to perform infrastructure-related operations (e.g. looking up the topology of a stream to find an appropriate node to connect to).The environment uses the number of passed-in URIs to choose an appropriate default number and will pick 1 in this case, which may be too low for a cluster deployment.This is why it is recommended to set the value explicitly, 3 being a good default.

Streams are usually long-lived, centrally-managed entities, that is, applicationsare not supposed to create and delete them. It is nevertheless possible to create anddelete stream with theEnvironment. This comes in handy for development and testingpurposes.

Streams are created with theEnvironment#streamCreator() method:

StreamCreator#create is idempotent: trying to re-create a stream with the same nameand same properties (e.g. maximum size, see below) will not throw an exception. In other words,you can be sure the stream has been created onceStreamCreator#create returns. Note it isnot possible to create a stream with the same name as an existing stream but with differentproperties. Such a request will result in an exception.

Streams can be deleted with theEnvironment#delete(String) method:

Note you should avoid stream churn (creating and deleting streams repetitively)as their creation and deletion imply some significant housekeeping onthe server side (interactions with the file system, communication between nodes of the cluster).

It is also possible to limit the size of a streamwhen creating it. A streamis an append-only data structure and reading from it does not remove data.This means a stream can grow indefinitely. RabbitMQ Stream supports asize-based and time-based retention policies: once the stream reaches a given sizeor a given age, it is truncated (starting from the beginning).

It is possible to set up the retention policy when creating the stream:

The previous snippet mentions a segment size. RabbitMQ Stream does not store a streamin a big, single file, it uses segment files for technical reasons.A stream is truncated by deleting whole segment files (and not part of them)sothe maximum size of a stream is usually significantly higher than the size ofsegment files. 500 MB is a reasonable segment file size to begin with.

RabbitMQ Stream also supports a time-based retention policy: segments get truncatedwhen they reach a certain age. The following snippetillustrates how to set the time-based retention policy:

AProducer instance is created from theEnvironment. The only mandatorysetting to specify is the stream to publish to:

TreatProducer instances as long-lived objects.Avoid creating a producer for single-message operations.

Internally, theEnvironment will query the broker to find out aboutthe topology of the stream and will create or re-use a connection topublish to the leader node of the stream.

The following table sums up the main settings to create aProducer:

Once aProducer has been created, it is possible to send a message withtheProducer#send(Message, ConfirmationHandler) method. The followingsnippet shows how to publish a message with a byte array payload:

Messages are not only made of abyte[] payload, as shown inthe next section they can also carry pre-defined and application properties.

TheConfirmationHandler defines an asynchronous callback invoked when the broker confirms message receipt.TheConfirmationHandler is the place for any logic on publishing confirmation, including re-publishing the message if it is negatively acknowledged.

The publishing example above showed that messages are made ofa byte array payload, but it did not go much further. Messages in RabbitMQ Streamcan actually be more sophisticated, as they comply with theAMQP 1.0 message format.

In a nutshell, a message in RabbitMQ Stream has the following structure:

• properties:a defined set of standard properties of the message (e.g.message ID, correlation ID, content type, etc).

• application properties: a set of arbitrary key/value pairs.

• body: typically an array of bytes.

• message annotations: a set of key/value pairs (aimed at the infrastructure).

The RabbitMQ Stream Java client uses theMessage interface to abstracta message and the recommended way to createMessage instances is touse theProducer#messageBuilder() method. To publish aMessage, usetheProducer#send(Message,ConfirmationHandler):

RabbitMQ Stream provides publisher confirms to avoid losing messages: oncethe broker has persisted a message it sends a confirmation for this message.But this can lead to duplicate messages: imagine the connection closesbecause of a network glitch after the message has been persisted butbeforethe confirmation reaches the producer. Once reconnected, the producer willretry to send the same message, as it never received the confirmation. So themessage will be persisted twice.

Luckily RabbitMQ Stream can detect and filter out duplicated messages, basedon 2 client-side elements: theproducer name and themessage publishing ID.

Setting the Name of a Producer

Producer producer = environment.producerBuilder()    .name("my-app-producer")(1)    .confirmTimeout(Duration.ZERO)(2)    .stream("my-stream")    .build();

Understanding Publishing ID

Message message = producer.messageBuilder()    .publishingId(1)(1)    .addData("hello".getBytes(StandardCharsets.UTF_8))    .build();producer.send(message, confirmationStatus -> { });

Restarting a Producer Where It Left Off

Producer producer = environment.producerBuilder()    .name("my-app-producer")(1)    .confirmTimeout(Duration.ZERO)(2)    .stream("my-stream")    .build();long nextPublishingId = producer.getLastPublishingId() +1;(3)while (moreContent(nextPublishingId)) {byte[] content = getContent(nextPublishingId);(4)    Message message = producer.messageBuilder()        .publishingId(nextPublishingId)(5)        .addData(content)        .build();    producer.send(message, confirmationStatus -> {});    nextPublishingId++;}

RabbitMQ Stream provides a special mode to publish, store, and dispatch messages: sub-entry batching.This mode increases throughput at the cost of increased latency and potential duplicated messages even when deduplication is enabled.It also allows using compression to reduce bandwidth and storage if messages are reasonably similar, at the cost of increasing CPU usage on the client side.

Sub-entry batching consists in squeezing several messages – a batch – in the slot that is usually used for one message.This means outbound messages are not only batched in publishing frames, but in sub-entries as well.

You can enable sub-entry batching by setting theProducerBuilder#subEntrySize parameter to a value greater than 1, like in the following snippet:

Reasonable values for the sub-entry size usually go from 10 to a few dozens.

A sub-entry batch will go directly to disc after it reached the broker, so the publishing client has complete control over it.This is the occasion to take advantage of the similarity of messages and compress them.There is no compression by default but you can choose among several algorithms with theProducerBuilder#compression(Compression) method:

Note the messages in a sub-entry are compressed altogether to benefit from their potential similarity, not one by one.

The following table lists the supported algorithms, general information about them, and the respective implementations used by default.

You are encouraged to test and evaluate the compression algorithms depending on your needs.

The compression libraries are pluggable thanks to theEnvironmentBuilder#compressionCodecFactory(CompressionCodecFactory) method.

AConsumer instance is created withEnvironment#consumerBuilder(). The mainsettings are the stream to consume from, the place in the stream to startconsuming from (theoffset), and a callback when a message is received(theMessageHandler). The next snippet shows how to create aConsumer:

The broker starts sending messages as soon as theConsumer instance is created.

The following table sums up the main settings to create aConsumer:

The offset is the place in the stream where the consumer starts consuming from.The possible values for the offset parameter are the following:

• OffsetSpecification.first(): starting from the first available offset. Ifthe stream has not beentruncated, thismeans the beginning of the stream (offset 0).

• OffsetSpecification.last(): starting from the end of the stream and returningthe lastchunk of messages immediately (if the stream is not empty).

• OffsetSpecification.next(): starting from the next offset to be written. ContrarytoOffsetSpecification.last(), consuming withOffsetSpecification.next()will not return anything if no-one is publishing to the stream. The broker will startsending messages to the consumer when messages are published to the stream.

• OffsetSpecification.offset(offset): starting from the specified offset. 0 means consumingfrom the beginning of the stream (first messages). The clientcan also specify any number, for example the offset where it left offin a previous incarnation of the application.

• OffsetSpecification.timestamp(timestamp): starting from the messages storedafter the specified timestamp.Note consumers can receive messages published a bit before the specified timestamp.Application code can filter out those messages if necessary.

The following figure shows the different offset specifications in a streammade of 2 chunks:

Each chunk contains a timestamp of its creation time.The broker uses this timestamp to find the appropriate chunk to start from when using a timestamp specification.The broker chooses the closest chunkbefore the specified timestamp, that is why consumers may see messages published a bit before what they specified.

RabbitMQ Stream provides server-side offset tracking.This means a consumer can track the offset it has reached in a stream.It allows a new incarnation of the consumer to restart consuming where it left off.All of this without an extra datastore, as the broker stores the offset tracking information.

Offset tracking works in 2 steps:

• the consumer must have aname. The name is set withConsumerBuilder#name(String). The namecan be any value (under 256 characters) and is expected to be unique (from the applicationpoint of view). Note neither the client library, nor the brokerenforces uniqueness of the name: if 2Consumer Java instances share the same name, theiroffset tracking will likely be interleaved, which applications usually do not expect.

• the consumer must periodicallystore the offset it has reached so far. The wayoffsets are stored depends on the tracking strategy: automatic or manual.

Whatever tracking strategy you use,a consumer must have a name to be able to store offsets.

Automatic Offset Tracking

Consumer consumer =    environment.consumerBuilder()        .stream("my-stream")        .name("application-1")(1)        .autoTrackingStrategy()(2)        .builder()        .messageHandler((context, message) -> {// message handling code...        })        .build();

Consumer consumer =    environment.consumerBuilder()        .stream("my-stream")        .name("application-1")(1)        .autoTrackingStrategy()(2)            .messageCountBeforeStorage(50_000)(3)            .flushInterval(Duration.ofSeconds(10))(4)        .builder()        .messageHandler((context, message) -> {// message handling code...        })        .build();

Consumer consumer =    environment.consumerBuilder()        .stream("my-stream")        .name("application-1")(1)        .messageHandler((context, message) -> {// message handling code...        })        .build();

Manual Offset Tracking

Consumer consumer =    environment.consumerBuilder()        .stream("my-stream")        .name("application-1")(1)        .manualTrackingStrategy()(2)        .builder()        .messageHandler((context, message) -> {// message handling code...if (conditionToStore()) {            context.storeOffset();(3)          }        })        .build();

Consumer consumer =    environment.consumerBuilder()        .stream("my-stream")        .name("application-1")(1)        .manualTrackingStrategy()(2)            .checkInterval(Duration.ofSeconds(10))(3)        .builder()        .messageHandler((context, message) -> {// message handling code...if (conditionToStore()) {            context.storeOffset();(4)          }        })        .build();

Considerations On Offset Tracking

Subscription Listener

Consumer consumer = environment.consumerBuilder()    .stream("my-stream")    .subscriptionListener(subscriptionContext -> {(1)long offset = getOffsetFromExternalStore();(2)        subscriptionContext.offsetSpecification(OffsetSpecification.offset(offset +1));(3)    })    .messageHandler((context, message) -> {// message handling code...        storeOffsetInExternalStore(context.offset());(4)    })    .build();

This section covers how a consumer can tell the broker when to send more messages.

By default, the broker keeps sending messages as long as messages are processed and theMessageHandler#handle(Context, Message) method returns.This strategy works fine if message processing is fast enough.If message processing takes longer, one can be tempted to process messages in parallel with anExecutorService.This will make thehandle method return immediately and the broker will keep sending messages, potentially overflowing the consumer.

What we miss in the parallel processing case is a way to tell the library we are done processing a message and that we are ready at some point to handle more messages.This is the goal of theMessageHandler.Context#processed() method.

This method is by default a no-op because the default flow control strategy keeps asking for more messages as soon as message processing is done.This method gets some real behavior to control the flow of messages when an appropriateConsumerFlowStrategy is setConsumerBuilder#flow().The following code snippet shows how to set a handy consumer flow strategy:

In the example we set up thecreditWhenHalfMessagesProcessed strategy which asks for more messages once half of the current messages have been marked as processed.The broker does not send messages one by one, it sendschunks of messages.A chunk of messages can contain 1 to several thousands of messages.So with the strategy set above, onceprocessed() has been called for half of the messages of the current chunk, the library will ask the broker for another one (it will provide acredit for the subscription).By doing this, the next chunk should arrive by the time we are done with the other half of the current chunk.This way the consumer is neither overwhelmed nor idle.

TheConsumerFlowStrategy interface provides some static helpers to configure the appropriate strategy.

Additional notes on consumer flow control:

• Make sure tocall theprocessed() method once you set up aConsumerFlowStrategy.The method is a no-op by default, but it is essential to call it with count-based strategies likecreditWhenHalfMessagesProcessed orcreditOnProcessedMessageCount.No calling it will stop the dispatching of messages.

• Make sure to callprocessed() only once.Whether the method is idempotent depends on the flow strategy implementation.Apart from the default one, the implementations the library provides does not makeprocessed() idempotent.

When the single active consumer feature is enabled for several consumer instances sharing the same stream and name, only one of these instances will be active at a time and so will receive messages.The other instances will be idle.

The single active consumer feature provides 2 benefits:

• Messages are processed in order: there is only one consumer at a time.

• Consumption continuity is maintained: a consumer from the group will take over if the active one stops or crashes.

A typical sequence of events would be the following:

• Several instances of the same consuming application start up.

• Each application instance registers a single active consumer.The consumer instances share the same name.

• The broker makes the first registered consumer the active one.

• The active consumer receives and processes messages, the other consumer instances remain idle.

• The active consumer stops or crashes.

• The broker chooses the consumer next in line to become the new active one.

• The new active consumer starts receiving messages.

The next figures illustrates this mechanism.There can be only one active consumer:

The broker rolls over to another consumer when the active one stops or crashes:

Note there can be several groups of single active consumers on the same stream.What makes them different from each other is the name used by the consumers.The broker deals with them independently.Let’s use an example.Imagine 2 differentapp-1 andapp-2 applications consuming from the same stream, with 3 identical instances each.Each instance registers 1 single active consumer with the name of the application.We end up with 3app-1 consumers and 3app-2 consumers, 1 active consumer in each group, so overall 6 consumers and 2 active ones, all of this on the same stream.

Let’s see now the API for single active consumer.

Enabling Single Active Consumer

Consumer consumer = environment.consumerBuilder()    .stream("my-stream")    .name("application-1")(1)    .singleActiveConsumer()(2)    .messageHandler((context, message) -> {// message handling code...    })    .build();

Offset Tracking

Reacting to Consumer State Change

Consumer consumer = environment.consumerBuilder()    .stream("my-stream")    .name("application-1")(1)    .singleActiveConsumer()(2)    .noTrackingStrategy()(3)    .consumerUpdateListener(context -> {(4)long offset = getOffsetFromExternalStore();(5)return OffsetSpecification.offset(offset +1);(6)    })    .messageHandler((context, message) -> {// message handling code...      storeOffsetInExternalStore(context.offset());    })    .build();

The topology of a super stream follows theAMQP 0.9.1 model: exchanges, queues, and bindings.AMQP resources are not used to transport or store stream messages.Instead, they describe the super stream topology and define which streams compose the super stream.

Let’s take the example of aninvoices super stream made of 3 streams (i.e. partitions):

• aninvoices exchange represents the super stream

• theinvoices-0,invoices-1,invoices-2 streams are the partitions of the super stream (streams are also AMQP queues in RabbitMQ)

• 3 bindings between the exchange and the streams link the super stream to its partitions and representrouting rules

When a super stream is in use, the stream Java client queries this information to find out about the partitions of a super stream and the routing rules.From the application code point of view, using a super stream is mostly configuration-based.Some logic must also be provided to extract routing information from messages.

It is possible to manage super streams with

• the stream Java client, by usingEnvironment#streamCreator() andEnvironment#deleteSuperStream(String)

• theadd_super_stream anddelete_super_stream commands inrabbitmq-streams (CLI)

• any AMQP 0.9.1 client library

• themanagement plugin

The stream Java client and the dedicated CLI commands are easier to use as they take care of the topology details (exchange, streams, and bindings).

With the Client Library

environment.streamCreator().name("invoices")    .superStream()    .partitions(5).creator()    .create();

environment.streamCreator().name("invoices")    .superStream()    .bindingKeys("amer","emea","apac").creator()    .create();

With the CLI

rabbitmq-streams add_super_stream invoices --partitions 5

When the topology of a super stream like the one described above has been set, creating a producer for it is straightforward:

Although theinvoices super stream is not a physical stream, you must use its name when declaring the producer.The client automatically discovers the individual streams that compose the super stream.Your application code must provide logic to extract a routing key from each message using aFunction<Message, String>.The client hashes this routing key to determine the target stream using the partition list and a modulo operation.

The client uses 32-bitMurmurHash3 by default to hash the routing key.This hash function provides good uniformity, performance, and portability, making it a good default choice, but it is possible to specify a custom hash function:

Note using Java’shashCode() method is a debatable choice as potential producers in other languages are unlikely to implement it, making the routing different between producers in different languages.

Resolving Routes with Bindings

Figure 5. A super stream with a partition for a region in a world

Producer producer = environment.producerBuilder()    .superStream("invoices")    .routing(msg -> msg.getApplicationProperties().get("region").toString())(1)    .key()(2)    .producerBuilder()    .build();

Using a Custom Routing Strategy

publicinterfaceRoutingStrategy {/** Where to route a message. */List<String> route(Message message, Metadata metadata);/** Metadata on the super stream. */interfaceMetadata {List<String> partitions();List<String> route(String routingKey);  }}

AtomicLong messageCount =newAtomicLong(0);RoutingStrategy routingStrategy = (message, metadata) -> {List<String> partitions = metadata.partitions();String stream = partitions.get(        (int) messageCount.getAndIncrement() % partitions.size()    );returnCollections.singletonList(stream);};Producer producer = environment.producerBuilder()    .superStream("invoices")    .routing(null)(1)    .strategy(routingStrategy)(2)    .producerBuilder()    .build();

Deduplication

A super stream consumer is a composite consumer: it looks up the super stream partitions and creates a consumer for each of them.The programming model is the same as with regular consumers for the application developer: their main job is to provide the application code to process messages, that is aMessageHandler instance.The configuration is different though and this section covers its subtleties.But let’s focus on the behavior of a super stream consumer first.

Super Stream Consumer in Practice

Declaring a Super Stream Consumer

Consumer consumer = environment.consumerBuilder()    .superStream("invoices")(1)    .messageHandler((context, message) -> {// message processing    })    .build();// ...consumer.close();(2)

Offset Tracking

Single Active Consumer Support

Figure 6. A single active consumer on a super stream is a composite consumer that creates an individual consumer for each partition

Figure 7. Consumer instances spread across the super stream partitions and are activated accordingly

Consumer consumer = environment.consumerBuilder()    .superStream("invoices")(1)    .name("application-1")(2)    .singleActiveConsumer()(3)    .messageHandler((context, message) -> {// message processing    })    .build();// ...

Consumer consumer =    environment.consumerBuilder()        .superStream("invoices")(1)        .name("application-1")(2)        .singleActiveConsumer()(3)        .manualTrackingStrategy()(4)        .builder()        .consumerUpdateListener(context -> {(5)if(context.isActive()) {(6)try {return OffsetSpecification.offset(                       context.consumer().storedOffset() +1                   );               }catch (NoOffsetException e) {return OffsetSpecification.next();               }           }else {               context.consumer().store(lastProcessedOffsetForThisStream);(7)returnnull;           }        })        .messageHandler((context, message) -> {// message handling code...if (conditionToStore()) {                context.storeOffset();(8)            }        })        .build();// ...

Consumer consumer = environment.consumerBuilder()    .superStream("invoices")(1)    .name("application-1")(2)    .singleActiveConsumer()(3)    .noTrackingStrategy()(4)    .consumerUpdateListener(context -> {(5)if (context.isActive()) {(6)long offset = getOffsetFromExternalStore();return OffsetSpecification.offset(offset +1);      }returnnull;(7)    })    .messageHandler((context, message) -> {// message handling code...      storeOffsetInExternalStore(context.stream(), context.offset());(8)    })    .build();

Publishers must define logic to extract filter values from messages.The following snippet shows how to extract the filter value from an application property:

Filter values can be null, resulting inunfiltered messages that are published normally.

A consumer needs to set up one or several filter values and some filtering logic to enable filtering.The filtering logic must be consistent with the filter values.In the next snippet, the consumer wants to process only messages from the state of California.It sets a filter value tocalifornia and a predicate that accepts a message only if thestate application property iscalifornia:

The filter logic is aPredicate<Message>.It must returntrue if a message is accepted, following the same semantics asjava.util.stream.Stream#filter(Predicate).

As stated above, not all messages must have an associated filter value.Many applications may not need filtering, so they can publish messages the regular way.So a stream can contain messages with and without an associated filter value.

By default, messages without a filter value (a.k.aunfiltered messages) are not sent to a consumer that enabled filtering.

But what if a consumer wants to process messages with a filter value and messages without any filter value as well?It must use thematchUnfiltered() method in its declaration and also make sure to keep the filtering logic consistent:

In the example above, the filtering logic allows bothcalifornia messagesand messages without a state set as well.

Since the server may send non-matching messages due to the probabilistic nature of Bloom filters, the client-side filtering logic must be robust to avoid processing unwanted messages.

Good filter value candidates:

• Shared categorical values: geographical locations (countries, states), document types (payslip, invoice, order), product categories (book, luggage, toy)

• Values with reasonable cardinality (few to few thousand distinct values)

Poor filter value candidates:

• Unique identifiers (message IDs, timestamps, UUIDs)

• Values with extreme cardinality (tens of thousands of distinct values)

Environment env = Environment.builder()    .oauth2()(1)    .tokenEndpointUri("https://localhost:8443/uaa/oauth/token/")(2)    .clientId("rabbitmq").clientSecret("rabbitmq")(3)    .grantType("password")(4)    .parameter("username","rabbit_super")(5)    .parameter("password","rabbit_super")(5)    .sslContext(sslContext)(6)    .environmentBuilder()    .build();

<dependencies><dependency><groupId>io.netty</groupId><artifactId>netty-transport-native-epoll</artifactId><version>4.2.7.Final</version><classifier>linux-x86_64</classifier></dependency></dependencies>

dependencies {  compile"io.netty:netty-transport-native-epoll:4.2.7.Final:linux-x86_64"}

EventLoopGroup epollEventLoopGroup =new MultiThreadIoEventLoopGroup((1)    EpollIoHandler.newFactory()(1));(1)Environment environment = Environment.builder()    .netty()(2)        .eventLoopGroup(epollEventLoopGroup)(3)        .bootstrapCustomizer(b -> b.channel(EpollSocketChannel.class))(4)        .environmentBuilder()    .build();

Observation for processing a message.

Span namerabbitmq.stream.process (defined by convention classcom.rabbitmq.stream.observation.micrometer.DefaultProcessObservationConvention).

Fully qualified name of the enclosing classcom.rabbitmq.stream.observation.micrometer.StreamObservationDocumentation.

Observation for publishing a message.

Span namerabbitmq.stream.publish (defined by convention classcom.rabbitmq.stream.observation.micrometer.DefaultPublishObservationConvention).

Fully qualified name of the enclosing classcom.rabbitmq.stream.observation.micrometer.StreamObservationDocumentation.

Observation for processing a message.

Metric namerabbitmq.stream.process (defined by convention classcom.rabbitmq.stream.observation.micrometer.DefaultProcessObservationConvention).Typetimer.

Metric namerabbitmq.stream.process.active (defined by convention classcom.rabbitmq.stream.observation.micrometer.DefaultProcessObservationConvention).Typelong task timer.

Fully qualified name of the enclosing classcom.rabbitmq.stream.observation.micrometer.StreamObservationDocumentation.

Observation for publishing a message.

Metric namerabbitmq.stream.publish (defined by convention classcom.rabbitmq.stream.observation.micrometer.DefaultPublishObservationConvention).Typetimer.

Metric namerabbitmq.stream.publish.active (defined by convention classcom.rabbitmq.stream.observation.micrometer.DefaultPublishObservationConvention).Typelong task timer.

Fully qualified name of the enclosing classcom.rabbitmq.stream.observation.micrometer.StreamObservationDocumentation.