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 is created. By default, the subscription is created at the end of the topic. See subscriptionInitialPosition(SubscriptionInitialPosition) to configure the initial position behavior.

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

      Returns:
      the consumer builder instance
      Throws:
      PulsarClientException - if 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 is created. By default, the subscription is created at the end of the topic. See subscriptionInitialPosition(SubscriptionInitialPosition) to configure the initial position behavior.

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

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

    • topic

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

    • topics

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

    • topicsPattern

      ConsumerBuilder<T> topicsPattern(Pattern topicsPattern)
      Specify a pattern for topics that this consumer subscribes to.

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

      The consumer automatically subscribes 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 subscribes to.

      It accepts a regular expression that is compiled into a pattern internally. E.g., "persistent://public/default/pattern-topic-.*"

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

      The consumer automatically subscribes 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 - the properties of the subscription
      Returns:
      the consumer builder instance

    • ackTimeout

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

      By default, the acknowledgment timeout is disabled (set to `0`, which means infinite). When a consumer with an infinite acknowledgment timeout terminates, any unacknowledged messages that it receives are re-delivered to another consumer.

      Since 2.3.0, when a dead letter policy is specified and no ackTimeoutMillis is specified, the acknowledgment timeout is set to 30 seconds.

      When enabling acknowledgment timeout, if a message is not acknowledged within the specified timeout, it is re-delivered to the consumer (possibly to a different consumer, in the 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)
      Acknowledgement returns receipt, but the message is not re-sent after getting receipt. Configure the acknowledgement timeout mechanism to redeliver the message if it is not acknowledged after ackTimeout, or to execute a timer task to check the acknowledgement timeout messages during every ackTimeoutTickTime period.
      Parameters:
      isAckReceiptEnabled - Boolean enables acknowledgement 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 a higher tick time reduces the memory overhead to track messages when the ack-timeout is set to bigger values (e.g., 1 hour).

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

    • negativeAckRedeliveryDelay

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

      When application uses Consumer.negativeAcknowledge(Message), the failed message is 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 a topic.

      Options are:

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

    • subscriptionMode

      ConsumerBuilder<T> subscriptionMode(SubscriptionMode subscriptionMode)
      Selects the subscription mode to be used when subscribing to a 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.

      The application receives messages through the message listener, and calls to Consumer.receive() are not 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 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 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 messages 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 takes 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 can potentially increase 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 subscriptions 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 a consumer receives a batch-message, it closes the consumer connection with the broker and Consumer.receive() calls remain blocked while Consumer.receiveAsync() receives exception in callback. The consumer is not able to receive any further messages unless batch-message in pipeline is removed.
      The default value is 1000 messages and should be adequate 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)
      Sets amount of time for group consumer acknowledgments.

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

      Setting a group time of 0 sends out acknowledgments immediately. A longer acknowledgment group time is more efficient, but at the expense of a slight increase in message re-deliveries after a failure.

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

    • maxAcknowledgmentGroupSize

      ConsumerBuilder<T> maxAcknowledgmentGroupSize(int messageNum)
      Set the number of messages for group consumer acknowledgments.

      By default, the consumer uses at most 1000 messages to send out acknowledgments to the broker.

      Parameters:
      messageNum -
      Returns:
      the consumer builder instance

    • replicateSubscriptionState

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

    • maxTotalReceiverQueueSizeAcrossPartitions

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

      This setting is 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)
      Sets the consumer name.

      Consumer names are informative, and can be used to identify 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 changes in a consumer group for failover subscriptions. The 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 reads messages from the compacted topic rather than the full message topic backlog. 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 are sent as normal.

      readCompacted can only be enabled on subscriptions to persistent topics with a single active consumer (i.e. failover or exclusive subscriptions). Enabling readCompacted on subscriptions to non-persistent topics or on shared subscriptions will cause the subscription call to throw a PulsarClientException.

      Parameters:
      readCompacted - whether to read from the compacted topic or full message topic backlog
      Returns:
      the consumer builder instance

    • patternAutoDiscoveryPeriod

      ConsumerBuilder<T> patternAutoDiscoveryPeriod(int periodInMinutes)
      Sets topic's auto-discovery period when using a pattern for topics consumer. The period is in minutes, and the default and minimum values are 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)
      Sets topic's 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 shared subscription consumers to determine which consumers the broker prioritizes when dispatching messages. Here, the broker follows descending priorities. (eg: 0=max-priority, 1, 2,..)

      In Shared subscription mode, the broker first dispatches messages to max priority-level consumers if they have permits, otherwise the broker considers next priority level consumers.

      If a subscription has consumer-A with priorityLevel 0 and Consumer-B with priorityLevel 1, then the broker dispatches messages to only consumer-A until it is drained, and then the broker will start 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 The broker selects the active consumer for a failover subscription based on consumer's priority-level and lexicographical sorting of 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)
      Sets a name/value property with this consumer.

      Properties are application-defined metadata that can be attached to the consumer. When getting topic stats, this metadata is associated with 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 topic stats, this metadata is associated with the consumer stats for easier identification.

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

    • subscriptionInitialPosition

      ConsumerBuilder<T> subscriptionInitialPosition(SubscriptionInitialPosition subscriptionInitialPosition)
      Sets 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 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)
      Sets dead letter policy for a consumer.

      By default, messages are redelivered as many times as possible until they are acknowledged. If you enable a dead letter mechanism, messages will have a maxRedeliverCount. When a message exceeds the maximum number of redeliveries, the message is sent to the Dead Letter Topic and acknowledged automatically.

      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 set a custom dead letter topic name: 
       client.newConsumer()
          .deadLetterPolicy(DeadLetterPolicy
              .builder()
              .maxRedeliverCount(10)
              .deadLetterTopic("your-topic-name")
              .build())
          .subscribe();

    • autoUpdatePartitions

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

    • autoUpdatePartitionsInterval

      ConsumerBuilder<T> autoUpdatePartitionsInterval(int interval, TimeUnit unit)
      Sets 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)
      Sets KeyShared subscription policy for consumer.

      By default, KeyShared subscriptions use auto split hash ranges to maintain consumers. If you want to set a different KeyShared policy, set a policy by using one of the following examples:

      Sticky hash range policy

       client.newConsumer()
          .keySharedPolicy(KeySharedPolicy.stickyHashRange().ranges(Range.of(0, 10)))
          .subscribe();
      For details about sticky hash range policies, see KeySharedPolicy.KeySharedPolicySticky.

      Auto-split hash range policy

       client.newConsumer()
          .keySharedPolicy(KeySharedPolicy.autoSplitHashRange())
          .subscribe();
      For details about auto-split hash range policies, see KeySharedPolicy.KeySharedPolicyAutoSplit.
      Parameters:
      keySharedPolicy - The KeySharedPolicy to specify

    • startMessageIdInclusive

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

    • batchReceivePolicy

      ConsumerBuilder<T> batchReceivePolicy(BatchReceivePolicy batchReceivePolicy)
      Sets BatchReceivePolicy for the consumer. By default, consumer uses 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 auto-retries messages. Default: disabled.
      Parameters:
      retryEnable - whether to auto retry message

    • enableBatchIndexAcknowledgment

      ConsumerBuilder<T> enableBatchIndexAcknowledgment(boolean batchIndexAcknowledgmentEnabled)
      Enable or disable batch index acknowledgment. To enable this feature, ensure batch index acknowledgment is enabled on 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 the consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acknowledging if autoAckOldestChunkedMessageOnQueueFull is true, otherwise it marks them for redelivery.
      Parameters:
      autoAckOldestChunkedMessageOnQueueFull -
      Returns:

    • expireTimeOfIncompleteChunkedMessage

      ConsumerBuilder<T> expireTimeOfIncompleteChunkedMessage(long duration, TimeUnit unit)
      If the producer fails to publish all the chunks of a message, then the consumer can expire incomplete chunks if the consumer doesn't receive all chunks during the expiration period (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, it causes a memory leak. If an application attempts to use an already “released” message, it might experience undefined behavior (eg: memory corruption, deserialization error, etc.).

    • messagePayloadProcessor

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

    • negativeAckRedeliveryBackoff

      ConsumerBuilder<T> negativeAckRedeliveryBackoff(RedeliveryBackoff negativeAckRedeliveryBackoff)
      negativeAckRedeliveryBackoff doesn't work with `consumer.negativeAcknowledge(MessageId messageId)` because we are unable 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)
      redeliveryBackoff doesn't work with `consumer.negativeAcknowledge(MessageId messageId)` because we are unable 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)
      Starts 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().

    • autoScaledReceiverQueueSizeEnabled

      ConsumerBuilder<T> autoScaledReceiverQueueSizeEnabled(boolean enabled)
      If this is enabled, the consumer receiver queue size is initialized as a very small value, 1 by default, and will double itself until it reaches the value set by receiverQueueSize(int), if and only if:

      1) User calls receive() and there are no messages in receiver queue.

      2) The last message we put in the receiver queue took the last space available in receiver queue. This is disabled by default and currentReceiverQueueSize is initialized as maxReceiverQueueSize. The feature should be able to reduce client memory usage.

      Parameters:
      enabled - whether to enable AutoScaledReceiverQueueSize.

    • topicConfiguration

      TopicConsumerBuilder<T> topicConfiguration(String topicName)
      Configure topic specific options to override those set at the ConsumerBuilder level.
      Parameters:
      topicName - a topic name
      Returns:
      a TopicConsumerBuilder instance

    • topicConfiguration

      ConsumerBuilder<T> topicConfiguration(String topicName, Consumer<TopicConsumerBuilder<T>> builderConsumer)
      Configure topic specific options to override those set at the ConsumerBuilder level.
      Parameters:
      topicName - a topic name
      builderConsumer - a consumer to allow the configuration of the TopicConsumerBuilder instance

    • topicConfiguration

      TopicConsumerBuilder<T> topicConfiguration(Pattern topicsPattern)
      Configure topic specific options to override those set at the ConsumerBuilder level.
      Parameters:
      topicsPattern - a regular expression to match a topic name
      Returns:
      a TopicConsumerBuilder instance

    • topicConfiguration

      ConsumerBuilder<T> topicConfiguration(Pattern topicsPattern, Consumer<TopicConsumerBuilder<T>> builderConsumer)
      Configure topic specific options to override those set at the ConsumerBuilder level.
      Parameters:
      topicsPattern - a regular expression to match a topic name
      builderConsumer - a consumer to allow the configuration of the TopicConsumerBuilder instance