ConsumerBuilder (Pulsar Client Java API)

All Superinterfaces:

Cloneable
```
public interface ConsumerBuilder<T>
extends Cloneable
```
ConsumerBuilder is used to configure and create instances of Consumer.

Since:

2.0.0

See Also:

PulsarClient.newConsumer()

Method Summary

All Methods Instance Methods Abstract Methods
Modifier and Type	Method and Description
`ConsumerBuilder<T>`	`acknowledgmentGroupTime(long delay, TimeUnit unit)` Group the consumer acknowledgments for the specified time.
`ConsumerBuilder<T>`	`ackTimeout(long ackTimeout, TimeUnit timeUnit)` Set the timeout for unacked messages, truncated to the nearest millisecond.
`ConsumerBuilder<T>`	`clone()` Create a copy of the current consumer builder.
`ConsumerBuilder<T>`	`consumerEventListener(ConsumerEventListener consumerEventListener)` Sets a `ConsumerEventListener` for the consumer.
`ConsumerBuilder<T>`	`consumerName(String consumerName)` Set the consumer name.
`ConsumerBuilder<T>`	`cryptoFailureAction(ConsumerCryptoFailureAction action)` Sets the ConsumerCryptoFailureAction to the value specified
`ConsumerBuilder<T>`	`cryptoKeyReader(CryptoKeyReader cryptoKeyReader)` Sets a `CryptoKeyReader`
`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.
`ConsumerBuilder<T>`	`intercept(ConsumerInterceptor<T>... interceptors)` Intercept `Consumer`.
`ConsumerBuilder<T>`	`loadConf(Map<String,Object> config)` Load the configuration from provided `config` map.
`ConsumerBuilder<T>`	`maxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions)` Set the max total receiver queue size across partitons.
`ConsumerBuilder<T>`	`messageListener(MessageListener<T> messageListener)` Sets a `MessageListener` for the consumer
`ConsumerBuilder<T>`	`patternAutoDiscoveryPeriod(int periodInMinutes)` Set topics auto discovery period when using a pattern for topics consumer.
`ConsumerBuilder<T>`	`priorityLevel(int priorityLevel)` Sets priority level for the shared subscription consumers to which broker gives more priority while dispatching messages.
`ConsumerBuilder<T>`	`properties(Map<String,String> properties)` Add all the properties in the provided map
`ConsumerBuilder<T>`	`property(String key, String value)` Set a name/value property with this consumer.
`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.
`ConsumerBuilder<T>`	`receiverQueueSize(int receiverQueueSize)` Sets the size of the consumer receive queue.
`Consumer<T>`	`subscribe()` Finalize the `Consumer` creation by subscribing to the topic.
`CompletableFuture<Consumer<T>>`	`subscribeAsync()` Finalize the `Consumer` creation by subscribing to the topic in asynchronous mode.
`ConsumerBuilder<T>`	`subscriptionInitialPosition(SubscriptionInitialPosition subscriptionInitialPosition)` Set subscriptionInitialPosition for the consumer
`ConsumerBuilder<T>`	`subscriptionName(String subscriptionName)` Specify the subscription name for this consumer.
`ConsumerBuilder<T>`	`subscriptionTopicsMode(Mode mode)` Determines to which topics this consumer should be subscribed to - Persistent, Non-Persistent, or both.
`ConsumerBuilder<T>`	`subscriptionType(SubscriptionType subscriptionType)` Select the subscription type to be used when subscribing to the topic.
`ConsumerBuilder<T>`	`topic(String... topicNames)` Specify the topics this consumer will subscribe on.
`ConsumerBuilder<T>`	`topics(List<String> topicNames)` Specify a list of topics that this consumer will subscribe on.
`ConsumerBuilder<T>`	`topicsPattern(java.util.regex.Pattern topicsPattern)` Specify a pattern for topics that this consumer will subscribe on.
`ConsumerBuilder<T>`	`topicsPattern(String topicsPattern)` Specify a pattern for topics that this consumer will subscribe on.

- Method Detail
  - 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 builder = client.newConsumer() //
         .subscriptionName("my-subscription-name") //
         .subscriptionType(SubscriptionType.Shared) //
         .receiverQueueSize(10);

 Consumer consumer1 = builder.clone().topic(TOPIC_1).subscribe();
 Consumer consumer2 = builder.clone().topic(TOPIC_2).subscribe();
 
```
  - 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);

 ConsumerBuilder<byte[]> builder = ...;
 builder = builder.loadConf(config);

 Consumer<byte[]> consumer = builder.subscribe();
 
```
    Parameters:
    
    config - configuration to load
    
    Returns:
    
    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 and all messages published after the creation will be retained until acknowledged, even if the consumer is not connected.
    
    Returns:
    
    the Consumer 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 and all messages published after the creation will be retained until acknowledged, 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 -
  - topics
```
ConsumerBuilder<T> topics(List<String> topicNames)
```
    Specify a list of topics that this consumer will subscribe on.
    
    Parameters:
    
    topicNames -
  - topicsPattern
```
ConsumerBuilder<T> topicsPattern(java.util.regex.Pattern topicsPattern)
```
    Specify a pattern for topics that this consumer will subscribe on.
    
    Parameters:
    
    topicsPattern -
  - 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://prop/use/ns-abc/pattern-topic-.*"
    
    Parameters:
    
    topicsPattern - given regular expression for topics pattern
  - subscriptionName
```
ConsumerBuilder<T> subscriptionName(String subscriptionName)
```
    Specify the subscription name for this consumer.
    This argument is required when constructing the consumer.
    
    Parameters:
    
    subscriptionName -
  - 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 10 seconds.
    
    Parameters:
    
    ackTimeout - for unacked messages.
    
    timeUnit - unit in which the timeout is provided.
  - subscriptionType
```
ConsumerBuilder<T> subscriptionType(SubscriptionType subscriptionType)
```
    Select the subscription type to be used when subscribing to the topic.
    Default is SubscriptionType.Exclusive
    
    Parameters:
    
    subscriptionType - the subscription type value
  - 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
  - cryptoKeyReader
```
ConsumerBuilder<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader)
```
    Sets a CryptoKeyReader
    
    Parameters:
    
    cryptoKeyReader - CryptoKeyReader object
  - cryptoFailureAction
```
ConsumerBuilder<T> cryptoFailureAction(ConsumerCryptoFailureAction action)
```
    Sets the ConsumerCryptoFailureAction to the value specified
    
    Parameters:
    
    action - The consumer action
  - 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
  - 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.
    
    Parameters:
    
    delay - the max amount of time an acknowledgemnt can be delayed
    
    unit - the time unit for the delay
  - 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).
    
    Parameters:
    
    maxTotalReceiverQueueSizeAcrossPartitions -
  - consumerName
```
ConsumerBuilder<T> consumerName(String consumerName)
```
    Set the consumer name.
    
    Parameters:
    
    consumerName -
  - 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
  - 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
  - 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)
  - priorityLevel
```
ConsumerBuilder<T> priorityLevel(int priorityLevel)
```
    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
 
```
    Parameters:
    
    priorityLevel -
  - property
```
ConsumerBuilder<T> property(String key,
                            String value)
```
    Set a name/value property with this consumer.
    
    Parameters:
    
    key -
    
    value -
  - properties
```
ConsumerBuilder<T> properties(Map<String,String> properties)
```
    Add all the properties in the provided map
    
    Parameters:
    
    properties -
  - subscriptionInitialPosition
```
ConsumerBuilder<T> subscriptionInitialPosition(SubscriptionInitialPosition subscriptionInitialPosition)
```
    Set subscriptionInitialPosition for the consumer
  - subscriptionTopicsMode
```
ConsumerBuilder<T> subscriptionTopicsMode(Mode mode)
```
    Determines to which topics this consumer should be subscribed to - Persistent, Non-Persistent, or both. Only used with pattern subscriptions.
    
    Parameters:
    
    mode - 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

Interface ConsumerBuilder<T>

Method Summary

Method Detail

clone

loadConf

subscribe

subscribeAsync

topic

topics

topicsPattern

topicsPattern

subscriptionName

ackTimeout

subscriptionType

messageListener

cryptoKeyReader

cryptoFailureAction

receiverQueueSize

acknowledgmentGroupTime

maxTotalReceiverQueueSizeAcrossPartitions

consumerName

consumerEventListener

readCompacted

patternAutoDiscoveryPeriod

priorityLevel

property

properties

subscriptionInitialPosition

subscriptionTopicsMode

intercept

deadLetterPolicy