Interface EnvironmentBuilder

All Known Implementing Classes:
StreamEnvironmentBuilder
public interface EnvironmentBuilder
API to configure and create an Environment.
See Also:

  • Method Details

    • uri

      EnvironmentBuilder uri(String uri)
      The URI of a node to connect to.

      URI must be of the form rabbitmq-stream://guest:guest@localhost:5552/%2f.

      Parameters:
      uri -
      Returns:
      this builder instance

    • uris

      EnvironmentBuilder uris(List<String> uris)
      A list of URIs of nodes of the same cluster to use to connect to.

      URIs must be of the form rabbitmq-stream://guest:guest@localhost:5552/%2f.

      Parameters:
      uris -
      Returns:
      this builder instance

    • addressResolver

      EnvironmentBuilder addressResolver(AddressResolver addressResolver)
      An AddressResolver to potentially change resolved node address to connect to.

      Applications can use this abstraction to make sure connection attempts ignore metadata hints and always go to a single point like a load balancer. Consider setting locatorConnectionCount(int) when using a load balancer.

      The default implementation does not perform any logic, it just returns the passed-in address.

      The default implementation is overridden automatically if the following conditions are met: the host to connect to is localhost, the user is guest, and no address resolver has been provided. The client will then always try to connect to localhost to facilitate local development. Just provide a pass-through address resolver to avoid this behavior, e.g.: 

      Environment.builder()
  .addressResolver(address -> address)
  .build();
      Parameters:
      addressResolver - the address resolver
      Returns:
      this builder instance
      See Also:

    • host

      EnvironmentBuilder host(String host)
      The host to connect to.
      Parameters:
      host -
      Returns:
      this builder instance

    • port

      EnvironmentBuilder port(int port)
      The port to use to connect.
      Parameters:
      port -
      Returns:
      this builder instance

    • codec

      EnvironmentBuilder codec(Codec codec)
      The AMQP 1.0 codec used to encode and decode AMQP 1.0 messages.
      Parameters:
      codec -
      Returns:
      this builder instance

    • id

      EnvironmentBuilder id(String id)
      Informational ID for this environment instance.

      This is currently used as a prefix for connection names. The broker does not enforce any kind of uniqueness based on this property. Default to rabbitmq-stream.

      Parameters:
      id -
      Returns:
      this builder instance

    • compressionCodecFactory

      EnvironmentBuilder compressionCodecFactory(CompressionCodecFactory compressionCodecFactory)
      Compression codec factory to use for compression in sub-entry batching.
      Parameters:
      compressionCodecFactory -
      Returns:
      this builder instance
      See Also:

    • rpcTimeout

      EnvironmentBuilder rpcTimeout(Duration timeout)
      Timeout for RPC calls.

      Default is 10 seconds.

      Parameters:
      timeout -
      Returns:
      this builder instance

    • saslConfiguration

      EnvironmentBuilder saslConfiguration(SaslConfiguration saslConfiguration)
      The SASL configuration to use.
      Parameters:
      saslConfiguration -
      Returns:
      this builder instance
      See Also:

    • credentialsProvider

      EnvironmentBuilder credentialsProvider(CredentialsProvider credentialsProvider)
      The CredentialsProvider to use.
      Parameters:
      credentialsProvider -
      Returns:
      this builder instance
      See Also:

    • username

      EnvironmentBuilder username(String username)
      The username to use.
      Parameters:
      username -
      Returns:
      this builder instance

    • password

      EnvironmentBuilder password(String password)
      The password to use.
      Parameters:
      password -
      Returns:
      this builder instance

    • virtualHost

      EnvironmentBuilder virtualHost(String virtualHost)
      The virtual host to connect to.
      Parameters:
      virtualHost -
      Returns:
      this builder instance

    • requestedHeartbeat

      EnvironmentBuilder requestedHeartbeat(Duration requestedHeartbeat)
      The heartbeat to request.

      Default is 60 seconds.

      Parameters:
      requestedHeartbeat -
      Returns:
      this builder instance
      See Also:

    • requestedMaxFrameSize

      EnvironmentBuilder requestedMaxFrameSize(int requestedMaxFrameSize)
      The maximum frame size to request.

      Default is 1048576.

      Parameters:
      requestedMaxFrameSize -
      Returns:
      this builder instance
      See Also:

    • maxFrameSizeBeforeAuthentication

      EnvironmentBuilder maxFrameSizeBeforeAuthentication(int maxFrameSizeBeforeAuthentication)
      The maximum frame size accepted from the broker before authentication completes.

      Until the authentication sequence completes (peer properties, SASL handshake, SASL authenticate, tune, open), the client has not negotiated the actual max frame size with the broker yet, so it applies this lower ceiling instead. This avoids exposing the client to an unbounded or overly large frame from a peer it has not authenticated yet.

      Default is 8192, which matches the broker default (stream.initial_frame_max) and is unlikely to need to change: it is small enough to mitigate the risk mentioned above, while being large enough to accommodate realistic JWT tokens carried as the SASL PLAIN password. The broker should be configured with the same value.

      Parameters:
      maxFrameSizeBeforeAuthentication -
      Returns:
      this builder instance
      See Also:

    • maxUncompressedSubEntryBatchSize

      EnvironmentBuilder maxUncompressedSubEntryBatchSize(int maxUncompressedSubEntryBatchSize)
      The maximum uncompressed size of a single sub-entry batch.

      A sub-entry batch header is written by the publishing client, stored verbatim by the broker, and replayed to consumers. Consumers must decompress a batch before they can decode any of its records, so this setting bounds the heap a single frame can commit on the message dispatching thread.

      Default is 67108864 (64 MiB). The broker enforces the same limit when messages are published and should be configured with the same value: a batch larger than the broker's limit is rejected at publish time, whereas a batch larger than this client's limit but within the broker's is stored and then fails on every consumer.

      Parameters:
      maxUncompressedSubEntryBatchSize -
      Returns:
      this builder instance

    • maxUncompressedSizePerChunk

      EnvironmentBuilder maxUncompressedSizePerChunk(int maxUncompressedSizePerChunk)
      The maximum total uncompressed size of the sub-entry batches in a single chunk.

      A chunk can hold many sub-entry batches, so maxUncompressedSubEntryBatchSize(int) alone does not bound the aggregate decompression work a single frame can demand on the message dispatching thread. This setting bounds that aggregate.

      Default is 1073741824 (1 GiB).

      Parameters:
      maxUncompressedSizePerChunk -
      Returns:
      this builder instance

    • chunkChecksum

      EnvironmentBuilder chunkChecksum(ChunkChecksum chunkChecksum)
      The checksum strategy used for chunk checksum.

      The default is CRC32 based on JDK implementation.

      Parameters:
      chunkChecksum -
      Returns:
      this builder instance

    • clientProperties

      EnvironmentBuilder clientProperties(Map<String,String> clientProperties)
      Custom client properties to add to default client properties.
      Parameters:
      clientProperties -
      Returns:
      this builder instance

    • clientProperty

      EnvironmentBuilder clientProperty(String key, String value)
      Add a custom client properties to default client properties.
      Parameters:
      key -
      value -
      Returns:
      this builder instance

    • metricsCollector

      EnvironmentBuilder metricsCollector(MetricsCollector metricsCollector)
      Set up a MetricsCollector.
      Parameters:
      metricsCollector -
      Returns:
      this builder instance

    • observationCollector

      EnvironmentBuilder observationCollector(ObservationCollector<?> observationCollector)
      Set up an ObservationCollector.
      Parameters:
      observationCollector -
      Returns:
      this builder instance

    • maxProducersByConnection

      EnvironmentBuilder maxProducersByConnection(int maxProducersByConnection)
      The maximum number of producers allocated to a single connection.

      Default is 256, which is the maximum value.

      The limit may not be strictly enforced in case of too many concurrent creations.

      Parameters:
      maxProducersByConnection -
      Returns:
      this builder instance

    • maxTrackingConsumersByConnection

      EnvironmentBuilder maxTrackingConsumersByConnection(int maxTrackingConsumersByConnection)
      The maximum number of tracking consumers allocated to a single connection.

      Default is 50, which is the maximum value.

      The limit may not be strictly enforced in case of too many concurrent creations.

      Parameters:
      maxTrackingConsumersByConnection -
      Returns:
      this builder instance

    • maxConsumersByConnection

      EnvironmentBuilder maxConsumersByConnection(int maxConsumersByConnection)
      The maximum number of consumers allocated to a single connection.

      Default is 256, which is the maximum value.

      The limit may not be strictly enforced in case of too many concurrent creations.

      Parameters:
      maxConsumersByConnection -
      Returns:
      this builder instance

    • scheduledExecutorService

      EnvironmentBuilder scheduledExecutorService(ScheduledExecutorService scheduledExecutorService)
      Set the ScheduledExecutorService used to:
      • Schedule producers batch sending
      • Handle connection recovery
      • Handle topology update
      Parameters:
      scheduledExecutorService - the service to use
      Returns:
      this builder instance

    • recoveryBackOffDelayPolicy

      EnvironmentBuilder recoveryBackOffDelayPolicy(BackOffDelayPolicy recoveryBackOffDelayPolicy)
      The BackOffDelayPolicy to use for connection recovery.

      The default is a fixed delay of 5 seconds.

      Parameters:
      recoveryBackOffDelayPolicy -
      Returns:
      this builder instance

    • topologyUpdateBackOffDelayPolicy

      EnvironmentBuilder topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy topologyUpdateBackOffDelayPolicy)
      The BackOffDelayPolicy to use for topology recovery.

      Topology recovery can kick in when streams leaders and replicas move from nodes to nodes in the cluster.

      The default is a first delay of 5 seconds, then 1 second for the next attempts.

      Parameters:
      topologyUpdateBackOffDelayPolicy -
      Returns:
      this builder instance

    • lazyInitialization

      EnvironmentBuilder lazyInitialization(boolean lazy)
      To delay the connection opening until necessary.

      No connection will be open before it is necessary (for stream management or producer/consumer creation).

      Default is false.

      Parameters:
      lazy -
      Returns:
      this builder instance

    • forceReplicaForConsumers

      EnvironmentBuilder forceReplicaForConsumers(boolean forceReplica)
      Flag to force the connection to a stream replica for consumers.

      The library will always prefer to connect to a stream replica to consume from, but it will fall back to the stream leader if no replica is available. This is the default behavior. Set this flag to true to make the library wait for a replica to become available if only the stream leader is available. This can lead to longer recovery time but help to offload a stream leader and let it deal only with write requests.

      Note the library performs only 5 attempts to locate a replica before falling back to the leader when the flag is set to true.

      The recoveryBackOffDelayPolicy(BackOffDelayPolicy) and topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy) policies control the time between attempts.

      Do not set this flag to true when streams have only 1 member (the leader), e.g. for local development.

      Default is false.

      Parameters:
      forceReplica - whether to force the connection to a replica or not
      Returns:
      this builder instance
      Since:
      0.13.0
      See Also:

    • forceLeaderForProducers

      EnvironmentBuilder forceLeaderForProducers(boolean forceLeader)
      Flag to force the connection to the stream leader for producers.

      The library prefers to connect to a node that hosts a stream leader for producers (default behavior).

      When using a load balancer, the library does not know in advance the node it connects to. It may have to retry to connect to the appropriate node.

      It will retry until it connects to the appropriate node (flag set to true, the default). This provides the best data locality, but may require several attempts, delaying the creation or the recovery of producers. This usually suits high-throughput use cases.

      The library will accept the connection to a stream replica if the flag is set to false . This will speed up the creation/recovery of producers, but at the cost of network hops between cluster nodes when publishing messages because only a stream leader accepts writes. This is usually acceptable for low-throughput use cases.

      Changing the default value should only benefit systems where a load balancer sits between the client applications and the cluster nodes.

      Default is true.

      Parameters:
      forceLeader - whether to force the connection to the leader or not
      Returns:
      this builder instance
      Since:
      0.21.0
      See Also:

    • locatorConnectionCount

      StreamEnvironmentBuilder locatorConnectionCount(int locatorConnectionCount)
      Set the expected number of "locator" connections to maintain.

      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).

      It is recommended to maintain 2 to 3 locator connections. The environment uses the smaller of the number of passed-in URIs and 3 by default (see uris(List)).

      The number of locator connections should be explicitly set when a load balancer is used, as the environment cannot know the number of cluster nodes in this case (the only URI set is the one of the load balancer).

      Parameters:
      locatorConnectionCount - number of expected locator connections
      Returns:
      this builder instance
      Since:
      0.21.0
      See Also:

    • build

      Environment build()
      Create the Environment instance.
      Returns:
      the configured environment

    • tls

      EnvironmentBuilder.TlsConfiguration tls()
      Enable and configure TLS.
      Returns:
      the TLS configuration helper

    • netty

      EnvironmentBuilder.NettyConfiguration netty()
      Helper to configure netty.
      Returns:
      Netty configuration helper

    • oauth2

      EnvironmentBuilder.OAuth2Configuration oauth2()
      OAuth 2 settings.
      Returns:
      OAuth 2 settings
      Since:
      1.3.0
      See Also: