Package org.apache.pulsar.client.api

Interface ConsumerBuilder<T>

All Superinterfaces:
Cloneable
@Public @Stable public interface ConsumerBuilder<T> extends Cloneable
ConsumerBuilder is used to configure and create instances of Consumer.
Since:
2.0.0
See Also:

  • Method Details

    • clone

      ConsumerBuilder<T> clone()
      Create a copy of the current consumer builder.

      Cloning the builder can be used to share an incomplete configuration and specialize it multiple times. For example: 

      
 ConsumerBuilder<String> builder = client.newConsumer(Schema.STRING)
         .subscriptionName("my-subscription-name")
         .subscriptionType(SubscriptionType.Shared)
         .receiverQueueSize(10);

 Consumer<String> consumer1 = builder.clone().topic("my-topic-1").subscribe();
 Consumer<String> consumer2 = builder.clone().topic("my-topic-2").subscribe();
      Returns:
      a cloned consumer builder object

    • loadConf

      ConsumerBuilder<T> loadConf(Map<String,Object> config)
      Load the configuration from provided config map.

      Example: 

      
 Map<String, Object> config = new HashMap<>();
 config.put("ackTimeoutMillis", 1000);
 config.put("receiverQueueSize", 2000);

 Consumer<byte[]> builder = client.newConsumer()
              .loadConf(config)
              .subscribe();

 Consumer<byte[]> consumer = builder.subscribe();
      Parameters:
      config - configuration to load
      Returns:
      the consumer builder instance

    • subscribe

      Consumer<T> subscribe() throws PulsarClientException
      Finalize the Consumer creation by subscribing to the topic.

      If the subscription does not exist, a new subscription will be created. By default the subscription will be created at the end of the topic. See subscriptionInitialPosition(SubscriptionInitialPosition) to configure the initial position behavior.

      Once a subscription is created, it will retain the data and the subscription cursor even if the consumer is not connected.

      Returns:
      the consumer builder instance
      Throws:
      PulsarClientException - if the the subscribe operation fails

    • subscribeAsync

      CompletableFuture<Consumer<T>> subscribeAsync()
      Finalize the Consumer creation by subscribing to the topic in asynchronous mode.

      If the subscription does not exist, a new subscription will be created. By default the subscription will be created at the end of the topic. See subscriptionInitialPosition(SubscriptionInitialPosition) to configure the initial position behavior.

      Once a subscription is created, it will retain the data and the subscription cursor even if the consumer is not connected.

      Returns:
      a future that will yield a Consumer instance
      Throws:
      PulsarClientException - if the the subscribe operation fails

    • topic

      ConsumerBuilder<T> topic(String... topicNames)
      Specify the topics this consumer will subscribe on.
      Parameters:
      topicNames - a set of topic that the consumer will subscribe on
      Returns:
      the consumer builder instance

    • topics

      ConsumerBuilder<T> topics(List<String> topicNames)
      Specify a list of topics that this consumer will subscribe on.
      Parameters:
      topicNames - a list of topic that the consumer will subscribe on
      Returns:
      the consumer builder instance

    • topicsPattern

      ConsumerBuilder<T> topicsPattern(Pattern topicsPattern)
      Specify a pattern for topics that this consumer will subscribe on.

      The pattern will be applied to subscribe to all the topics, within a single namespace, that will match the pattern.

      The consumer will automatically subscribe to topics created after itself.

      Parameters:
      topicsPattern - a regular expression to select a list of topics to subscribe to
      Returns:
      the consumer builder instance

    • topicsPattern

      ConsumerBuilder<T> topicsPattern(String topicsPattern)
      Specify a pattern for topics that this consumer will subscribe on.

      It accepts regular expression and will be compiled into a pattern internally. Eg. "persistent://public/default/pattern-topic-.*"

      The pattern will be applied to subscribe to all the topics, within a single namespace, that will match the pattern.

      The consumer will automatically subscribe to topics created after itself.

      Parameters:
      topicsPattern - given regular expression for topics pattern
      Returns:
      the consumer builder instance

    • subscriptionName

      ConsumerBuilder<T> subscriptionName(String subscriptionName)
      Specify the subscription name for this consumer.

      This argument is required when constructing the consumer.

      Parameters:
      subscriptionName - the name of the subscription that this consumer should attach to
      Returns:
      the consumer builder instance

    • subscriptionProperties

      ConsumerBuilder<T> subscriptionProperties(Map<String,String> subscriptionProperties)
      Specify the subscription properties for this subscription. Properties are immutable, and consumers under the same subscription will fail to create a subscription if they use different properties.
      Parameters:
      subscriptionProperties -
      Returns:

    • ackTimeout

      ConsumerBuilder<T> ackTimeout(long ackTimeout, TimeUnit timeUnit)
      Set the timeout for unacked messages, truncated to the nearest millisecond. The timeout needs to be greater than 1 second.

      By default, the acknowledge timeout is disabled and that means that messages delivered to a consumer will not be re-delivered unless the consumer crashes.

      When enabling ack timeout, if a message is not acknowledged within the specified timeout it will be re-delivered to the consumer (possibly to a different consumer in case of a shared subscription).

      Parameters:
      ackTimeout - for unacked messages.
      timeUnit - unit in which the timeout is provided.
      Returns:
      the consumer builder instance

    • isAckReceiptEnabled

      ConsumerBuilder<T> isAckReceiptEnabled(boolean isAckReceiptEnabled)
      Ack will return receipt but does not mean that the message will not be resent after get receipt.
      Parameters:
      isAckReceiptEnabled - Boolean is enable ack for receipt
      Returns:
      the consumer builder instance

    • ackTimeoutTickTime

      ConsumerBuilder<T> ackTimeoutTickTime(long tickTime, TimeUnit timeUnit)
      Define the granularity of the ack-timeout redelivery.

      By default, the tick time is set to 1 second. Using an higher tick time will reduce the memory overhead to track messages when the ack-timeout is set to bigger values (eg: 1hour).

      Parameters:
      tickTime - the min precision for the ack timeout messages tracker
      timeUnit - unit in which the timeout is provided.
      Returns:
      the consumer builder instance

    • negativeAckRedeliveryDelay

      ConsumerBuilder<T> negativeAckRedeliveryDelay(long redeliveryDelay, TimeUnit timeUnit)
      Set the delay to wait before re-delivering messages that have failed to be process.

      When application uses Consumer.negativeAcknowledge(Message), the failed message will be redelivered after a fixed timeout. The default is 1 min.

      Parameters:
      redeliveryDelay - redelivery delay for failed messages
      timeUnit - unit in which the timeout is provided.
      Returns:
      the consumer builder instance
      See Also:

    • subscriptionType

      ConsumerBuilder<T> subscriptionType(SubscriptionType subscriptionType)
      Select the subscription type to be used when subscribing to the topic.

      Options are:

      Parameters:
      subscriptionType - the subscription type value
      Returns:
      the consumer builder instance

    • subscriptionMode

      ConsumerBuilder<T> subscriptionMode(SubscriptionMode subscriptionMode)
      Select the subscription mode to be used when subscribing to the topic.

      Options are:

      Parameters:
      subscriptionMode - the subscription mode value
      Returns:
      the consumer builder instance

    • messageListener

      ConsumerBuilder<T> messageListener(MessageListener<T> messageListener)
      Sets a MessageListener for the consumer

      When a MessageListener is set, application will receive messages through it. Calls to Consumer.receive() will not be allowed.

      Parameters:
      messageListener - the listener object
      Returns:
      the consumer builder instance

    • cryptoKeyReader

      ConsumerBuilder<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader)
      Sets a CryptoKeyReader.

      Configure the key reader to be used to decrypt the message payloads.

      Parameters:
      cryptoKeyReader - CryptoKeyReader object
      Returns:
      the consumer builder instance

    • defaultCryptoKeyReader

      ConsumerBuilder<T> defaultCryptoKeyReader(String privateKey)
      Sets the default implementation of CryptoKeyReader.

      Configure the key reader to be used to decrypt the message payloads.

      Parameters:
      privateKey - the private key that is always used to decrypt message payloads.
      Returns:
      the consumer builder instance
      Since:
      2.8.0

    • defaultCryptoKeyReader

      ConsumerBuilder<T> defaultCryptoKeyReader(Map<String,String> privateKeys)
      Sets the default implementation of CryptoKeyReader.

      Configure the key reader to be used to decrypt the message payloads.

      Parameters:
      privateKeys - the map of private key names and their URIs used to decrypt message payloads.
      Returns:
      the consumer builder instance
      Since:
      2.8.0

    • messageCrypto

      ConsumerBuilder<T> messageCrypto(MessageCrypto messageCrypto)
      Sets a MessageCrypto.

      Contains methods to encrypt/decrypt message for End to End Encryption.

      Parameters:
      messageCrypto - MessageCrypto object
      Returns:
      the consumer builder instance

    • cryptoFailureAction

      ConsumerBuilder<T> cryptoFailureAction(ConsumerCryptoFailureAction action)
      Sets the ConsumerCryptoFailureAction to the value specified.
      Parameters:
      action - the action the consumer will take in case of decryption failures
      Returns:
      the consumer builder instance

    • receiverQueueSize

      ConsumerBuilder<T> receiverQueueSize(int receiverQueueSize)
      Sets the size of the consumer receive queue.

      The consumer receive queue controls how many messages can be accumulated by the Consumer before the application calls Consumer.receive(). Using a higher value could potentially increase the consumer throughput at the expense of bigger memory utilization.

      Setting the consumer queue size as zero

      • Decreases the throughput of the consumer, by disabling pre-fetching of messages. This approach improves the message distribution on shared subscription, by pushing messages only to the consumers that are ready to process them. Neither Consumer.receive(int, TimeUnit) nor Partitioned Topics can be used if the consumer queue size is zero. Consumer.receive() function call should not be interrupted when the consumer queue size is zero.
      • Doesn't support Batch-Message: if consumer receives any batch-message then it closes consumer connection with broker and Consumer.receive() call will remain blocked while Consumer.receiveAsync() receives exception in callback. consumer will not be able receive any further message unless batch-message in pipeline is removed
      Default value is 1000 messages and should be good for most use cases.
      Parameters:
      receiverQueueSize - the new receiver queue size value
      Returns:
      the consumer builder instance

    • acknowledgmentGroupTime

      ConsumerBuilder<T> acknowledgmentGroupTime(long delay, TimeUnit unit)
      Group the consumer acknowledgments for the specified time.

      By default, the consumer will use a 100 ms grouping time to send out the acknowledgments to the broker.

      Setting a group time of 0, will send out the acknowledgments immediately. A longer ack group time will be more efficient at the expense of a slight increase in message re-deliveries after a failure.

      Parameters:
      delay - the max amount of time an acknowledgemnt can be delayed
      unit - the time unit for the delay
      Returns:
      the consumer builder instance

    • replicateSubscriptionState

      ConsumerBuilder<T> replicateSubscriptionState(boolean replicateSubscriptionState)
      Parameters:
      replicateSubscriptionState -

    • maxTotalReceiverQueueSizeAcrossPartitions

      ConsumerBuilder<T> maxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions)
      Set the max total receiver queue size across partitons.

      This setting will be used to reduce the receiver queue size for individual partitions receiverQueueSize(int) if the total exceeds this value (default: 50000). The purpose of this setting is to have an upper-limit on the number of messages that a consumer can be pushed at once from a broker, across all the partitions.

      Parameters:
      maxTotalReceiverQueueSizeAcrossPartitions - max pending messages across all the partitions
      Returns:
      the consumer builder instance

    • consumerName

      ConsumerBuilder<T> consumerName(String consumerName)
      Set the consumer name.

      Consumer name is informative and it can be used to indentify a particular consumer instance from the topic stats.

      Parameters:
      consumerName -
      Returns:
      the consumer builder instance

    • consumerEventListener

      ConsumerBuilder<T> consumerEventListener(ConsumerEventListener consumerEventListener)
      Sets a ConsumerEventListener for the consumer.

      The consumer group listener is used for receiving consumer state change in a consumer group for failover subscription. Application can then react to the consumer state changes.

      Parameters:
      consumerEventListener - the consumer group listener object
      Returns:
      the consumer builder instance

    • readCompacted

      ConsumerBuilder<T> readCompacted(boolean readCompacted)
      If enabled, the consumer will read messages from the compacted topic rather than reading the full message backlog of the topic. This means that, if the topic has been compacted, the consumer will only see the latest value for each key in the topic, up until the point in the topic message backlog that has been compacted. Beyond that point, the messages will be sent as normal.

      readCompacted can only be enabled subscriptions to persistent topics, which have a single active consumer (i.e. failure or exclusive subscriptions). Attempting to enable it on subscriptions to a non-persistent topics or on a shared subscription, will lead to the subscription call throwing a PulsarClientException.

      Parameters:
      readCompacted - whether to read from the compacted topic
      Returns:
      the consumer builder instance

    • patternAutoDiscoveryPeriod

      ConsumerBuilder<T> patternAutoDiscoveryPeriod(int periodInMinutes)
      Set topics auto discovery period when using a pattern for topics consumer. The period is in minute, and default and minimum value is 1 minute.
      Parameters:
      periodInMinutes - number of minutes between checks for new topics matching pattern set with topicsPattern(String)
      Returns:
      the consumer builder instance

    • patternAutoDiscoveryPeriod

      ConsumerBuilder<T> patternAutoDiscoveryPeriod(int interval, TimeUnit unit)
      Set topics auto discovery period when using a pattern for topics consumer.
      Parameters:
      interval - the amount of delay between checks for new topics matching pattern set with topicsPattern(String)
      unit - the unit of the topics auto discovery period
      Returns:
      the consumer builder instance

    • priorityLevel

      ConsumerBuilder<T> priorityLevel(int priorityLevel)
      Shared subscription Sets priority level for the shared subscription consumers to which broker gives more priority while dispatching messages. Here, broker follows descending priorities. (eg: 0=max-priority, 1, 2,..)

      In Shared subscription mode, broker will first dispatch messages to max priority-level consumers if they have permits, else broker will consider next priority level consumers.

      If subscription has consumer-A with priorityLevel 0 and Consumer-B with priorityLevel 1 then broker will dispatch messages to only consumer-A until it runs out permit and then broker starts dispatching messages to Consumer-B. 

       Consumer PriorityLevel Permits
 C1       0             2
 C2       0             1
 C3       0             1
 C4       1             2
 C5       1             1
 Order in which broker dispatches messages to consumers: C1, C2, C3, C1, C4, C5, C4

      Failover subscription Broker selects active consumer for a failover-subscription based on consumer's priority-level and lexicographical sorting of a consumer name. eg: 

       1. Active consumer = C1 : Same priority-level and lexicographical sorting
 Consumer PriorityLevel Name
 C1       0             aaa
 C2       0             bbb

 2. Active consumer = C2 : Consumer with highest priority
 Consumer PriorityLevel Name
 C1       1             aaa
 C2       0             bbb

 Partitioned-topics:
 Broker evenly assigns partitioned topics to highest priority consumers.
      Parameters:
      priorityLevel - the priority of this consumer
      Returns:
      the consumer builder instance

    • property

      ConsumerBuilder<T> property(String key, String value)
      Set a name/value property with this consumer.

      Properties are application defined metadata that can be attached to the consumer. When getting the topic stats, this metadata will be associated to the consumer stats for easier identification.

      Parameters:
      key - the property key
      value - the property value
      Returns:
      the consumer builder instance

    • properties

      ConsumerBuilder<T> properties(Map<String,String> properties)
      Add all the properties in the provided map to the consumer.

      Properties are application defined metadata that can be attached to the consumer. When getting the topic stats, this metadata will be associated to the consumer stats for easier identification.

      Parameters:
      properties - the map with properties
      Returns:
      the consumer builder instance

    • subscriptionInitialPosition

      ConsumerBuilder<T> subscriptionInitialPosition(SubscriptionInitialPosition subscriptionInitialPosition)
      Set the SubscriptionInitialPosition for the consumer.
      Parameters:
      subscriptionInitialPosition - the position where to initialize a newly created subscription
      Returns:
      the consumer builder instance

    • subscriptionTopicsMode

      ConsumerBuilder<T> subscriptionTopicsMode(RegexSubscriptionMode regexSubscriptionMode)
      Determines to which topics this consumer should be subscribed to - Persistent, Non-Persistent, or both. Only used with pattern subscriptions.
      Parameters:
      regexSubscriptionMode - Pattern subscription mode

    • intercept

      ConsumerBuilder<T> intercept(ConsumerInterceptor<T>... interceptors)
      Intercept Consumer.
      Parameters:
      interceptors - the list of interceptors to intercept the consumer created by this builder.

    • deadLetterPolicy

      ConsumerBuilder<T> deadLetterPolicy(DeadLetterPolicy deadLetterPolicy)
      Set dead letter policy for consumer.

      By default some message will redelivery so many times possible, even to the extent that it can be never stop. By using dead letter mechanism messages will has the max redelivery count, when message exceeding the maximum number of redeliveries, message will send to the Dead Letter Topic and acknowledged automatic.

      You can enable the dead letter mechanism by setting dead letter policy. example: 

       client.newConsumer()
          .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(10).build())
          .subscribe();
      Default dead letter topic name is {TopicName}-{Subscription}-DLQ. To setting a custom dead letter topic name 
       client.newConsumer()
          .deadLetterPolicy(DeadLetterPolicy
              .builder()
              .maxRedeliverCount(10)
              .deadLetterTopic("your-topic-name")
              .build())
          .subscribe();
      When a dead letter policy is specified, and no ackTimeoutMillis is specified, then the ack timeout will be set to 30000 millisecond.

    • autoUpdatePartitions

      ConsumerBuilder<T> autoUpdatePartitions(boolean autoUpdate)
      If enabled, the consumer will auto subscribe for partitions increasement. This is only for partitioned consumer.
      Parameters:
      autoUpdate - whether to auto update partition increasement

    • autoUpdatePartitionsInterval

      ConsumerBuilder<T> autoUpdatePartitionsInterval(int interval, TimeUnit unit)
      Set the interval of updating partitions (default: 1 minute). This only works if autoUpdatePartitions is enabled.
      Parameters:
      interval - the interval of updating partitions
      unit - the time unit of the interval.
      Returns:
      the consumer builder instance

    • keySharedPolicy

      ConsumerBuilder<T> keySharedPolicy(KeySharedPolicy keySharedPolicy)
      Set KeyShared subscription policy for consumer.

      By default, KeyShared subscription use auto split hash range to maintain consumers. If you want to set a different KeyShared policy, you can set by following example: 

       client.newConsumer()
          .keySharedPolicy(KeySharedPolicy.stickyHashRange().ranges(Range.of(0, 10)))
          .subscribe();
      Details about sticky hash range policy, please see KeySharedPolicy.KeySharedPolicySticky.

      Or 

       client.newConsumer()
          .keySharedPolicy(KeySharedPolicy.autoSplitHashRange())
          .subscribe();
      Details about auto split hash range policy, please see KeySharedPolicy.KeySharedPolicyAutoSplit.
      Parameters:
      keySharedPolicy - The KeySharedPolicy want to specify

    • startMessageIdInclusive

      ConsumerBuilder<T> startMessageIdInclusive()
      Set the consumer to include the given position of any reset operation like or Consumer#seek(MessageId).
      Returns:
      the consumer builder instance

    • batchReceivePolicy

      ConsumerBuilder<T> batchReceivePolicy(BatchReceivePolicy batchReceivePolicy)
      Set batch receive policy BatchReceivePolicy for consumer. By default, consumer will use BatchReceivePolicy.DEFAULT_POLICY as batch receive policy.

      Example: 

       client.newConsumer().batchReceivePolicy(BatchReceivePolicy.builder()
              .maxNumMessages(100)
              .maxNumBytes(5 * 1024 * 1024)
              .timeout(100, TimeUnit.MILLISECONDS)
              .build()).subscribe();

    • enableRetry

      ConsumerBuilder<T> enableRetry(boolean retryEnable)
      If enabled, the consumer will auto retry message. default unabled.
      Parameters:
      retryEnable - whether to auto retry message

    • enableBatchIndexAcknowledgment

      ConsumerBuilder<T> enableBatchIndexAcknowledgment(boolean batchIndexAcknowledgmentEnabled)
      Enable or disable the batch index acknowledgment. To enable this feature must ensure batch index acknowledgment feature is enabled at the broker side.

    • maxPendingChuckedMessage

      @Deprecated ConsumerBuilder<T> maxPendingChuckedMessage(int maxPendingChuckedMessage)
      Deprecated.
      use maxPendingChunkedMessage(int)
      Consumer buffers chunk messages into memory until it receives all the chunks of the original message. While consuming chunk-messages, chunks from same message might not be contiguous in the stream and they might be mixed with other messages' chunks. so, consumer has to maintain multiple buffers to manage chunks coming from different messages. This mainly happens when multiple publishers are publishing messages on the topic concurrently or publisher failed to publish all chunks of the messages. 
       eg: M1-C1, M2-C1, M1-C2, M2-C2
 Here, Messages M1-C1 and M1-C2 belong to original message M1, M2-C1 and M2-C2 messages belong to M2 message.
      Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it can be guarded by providing this @maxPendingChuckedMessage threshold. Once, consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acking or asking broker to redeliver later by marking it unacked. This behavior can be controlled by configuration: @autoAckOldestChunkedMessageOnQueueFull The default value is 10.
      Parameters:
      maxPendingChuckedMessage -
      Returns:

    • maxPendingChunkedMessage

      ConsumerBuilder<T> maxPendingChunkedMessage(int maxPendingChunkedMessage)
      Consumer buffers chunk messages into memory until it receives all the chunks of the original message. While consuming chunk-messages, chunks from same message might not be contiguous in the stream and they might be mixed with other messages' chunks. so, consumer has to maintain multiple buffers to manage chunks coming from different messages. This mainly happens when multiple publishers are publishing messages on the topic concurrently or publisher failed to publish all chunks of the messages. 
       eg: M1-C1, M2-C1, M1-C2, M2-C2
 Here, Messages M1-C1 and M1-C2 belong to original message M1, M2-C1 and M2-C2 messages belong to M2 message.
      Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it can be guarded by providing this @maxPendingChunkedMessage threshold. Once, consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acking or asking broker to redeliver later by marking it unacked. This behavior can be controlled by configuration: @autoAckOldestChunkedMessageOnQueueFull The default value is 10.
      Parameters:
      maxPendingChunkedMessage -
      Returns:

    • autoAckOldestChunkedMessageOnQueueFull

      ConsumerBuilder<T> autoAckOldestChunkedMessageOnQueueFull(boolean autoAckOldestChunkedMessageOnQueueFull)
      Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it can be guarded by providing this @maxPendingChunkedMessage threshold. Once, consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acking if autoAckOldestChunkedMessageOnQueueFull is true else it marks them for redelivery.
      Parameters:
      autoAckOldestChunkedMessageOnQueueFull -
      Returns:

    • expireTimeOfIncompleteChunkedMessage

      ConsumerBuilder<T> expireTimeOfIncompleteChunkedMessage(long duration, TimeUnit unit)
      If producer fails to publish all the chunks of a message then consumer can expire incomplete chunks if consumer won't be able to receive all chunks in expire times (default 1 minute).
      Parameters:
      duration -
      unit -
      Returns:

    • poolMessages

      ConsumerBuilder<T> poolMessages(boolean poolMessages)
      Enable pooling of messages and the underlying data buffers.

      When pooling is enabled, the application is responsible for calling Message.release() after the handling of every received message. If “release()” is not called on a received message, there will be a memory leak. If an application attempts to use and already “released” message, it might experience undefined behavior (eg: memory corruption, deserialization error, etc.).

    • messagePayloadProcessor

      ConsumerBuilder<T> messagePayloadProcessor(MessagePayloadProcessor payloadProcessor)
      If it's configured with a non-null value, the consumer will use the processor to process the payload, including decoding it to messages and triggering the listener. Default: null

    • negativeAckRedeliveryBackoff

      ConsumerBuilder<T> negativeAckRedeliveryBackoff(RedeliveryBackoff negativeAckRedeliveryBackoff)
      Notice: the negativeAckRedeliveryBackoff will not work with `consumer.negativeAcknowledge(MessageId messageId)` because we are not able to get the redelivery count from the message ID.

      Example: 

       client.newConsumer().negativeAckRedeliveryBackoff(ExponentialRedeliveryBackoff.builder()
              .minNackTimeMs(1000)
              .maxNackTimeMs(60 * 1000)
              .build()).subscribe();

    • ackTimeoutRedeliveryBackoff

      ConsumerBuilder<T> ackTimeoutRedeliveryBackoff(RedeliveryBackoff ackTimeoutRedeliveryBackoff)
      Notice: the redeliveryBackoff will not work with `consumer.negativeAcknowledge(MessageId messageId)` because we are not able to get the redelivery count from the message ID.

      Example: 

       client.newConsumer().ackTimeout(10, TimeUnit.SECOND)
              .ackTimeoutRedeliveryBackoff(ExponentialRedeliveryBackoff.builder()
              .minNackTimeMs(1000)
              .maxNackTimeMs(60 * 1000)
              .build()).subscribe();

    • startPaused

      ConsumerBuilder<T> startPaused(boolean paused)
      Start the consumer in a paused state. When enabled, the consumer does not immediately fetch messages when subscribe() is called. Instead, the consumer waits to fetch messages until Consumer.resume() is called.

      See also Consumer.pause().