org.apache.pulsar.client.api

Interface Consumer<T>

All Superinterfaces:

AutoCloseable, Closeable

public interface Consumer<T>
extends Closeable

An interface that abstracts behavior of Pulsar's consumer.

Method Summary

Modifier and Type	Method and Description
void	acknowledge(Message<?> message) Acknowledge the consumption of a single message
void	acknowledge(MessageId messageId) Acknowledge the consumption of a single message, identified by its MessageId
CompletableFuture<Void>	acknowledgeAsync(Message<?> message) Asynchronously acknowledge the consumption of a single message
CompletableFuture<Void>	acknowledgeAsync(MessageId messageId) Asynchronously acknowledge the consumption of a single message
void	acknowledgeCumulative(Message<?> message) Acknowledge the reception of all the messages in the stream up to (and including) the provided message.
void	acknowledgeCumulative(MessageId messageId) Acknowledge the reception of all the messages in the stream up to (and including) the provided message.
CompletableFuture<Void>	acknowledgeCumulativeAsync(Message<?> message) Asynchronously Acknowledge the reception of all the messages in the stream up to (and including) the provided message.
CompletableFuture<Void>	acknowledgeCumulativeAsync(MessageId messageId) Asynchronously Acknowledge the reception of all the messages in the stream up to (and including) the provided message.
void	close() Close the consumer and stop the broker to push more messages.
CompletableFuture<Void>	closeAsync() Asynchronously close the consumer and stop the broker to push more messages
String	getConsumerName() Get the name of consumer.
ConsumerStats	getStats() Get statistics for the consumer.
String	getSubscription() Get a subscription for the consumer
String	getTopic() Get a topic for the consumer
boolean	hasReachedEndOfTopic() Return true if the topic was terminated and this consumer has already consumed all the messages in the topic.
boolean	isConnected()
void	pause() Stop requesting new messages from the broker until resume() is called.
Message<T>	receive() Receives a single message.
Message<T>	receive(int timeout, TimeUnit unit) Receive a single message
CompletableFuture<Message<T>>	receiveAsync() Receive a single message
void	redeliverUnacknowledgedMessages() Redelivers all the unacknowledged messages.
void	resume() Resume requesting messages from the broker.
void	seek(MessageId messageId) Reset the subscription associated with this consumer to a specific message id.
CompletableFuture<Void>	seekAsync(MessageId messageId) Reset the subscription associated with this consumer to a specific message id.
void	unsubscribe() Unsubscribe the consumer
CompletableFuture<Void>	unsubscribeAsync() Asynchronously unsubscribe the consumer

Method Detail

getTopic

String getTopic()

Get a topic for the consumer

Returns:

topic for the consumer

getSubscription

String getSubscription()

Get a subscription for the consumer

Returns:

subscription for the consumer

unsubscribe

void unsubscribe()
          throws PulsarClientException

Unsubscribe the consumer

This call blocks until the consumer is unsubscribed.

Throws:

PulsarClientException

unsubscribeAsync

CompletableFuture<Void> unsubscribeAsync()

Asynchronously unsubscribe the consumer

Returns:

CompletableFuture for this operation

receive

Message<T> receive()
            throws PulsarClientException

Receives a single message.

This calls blocks until a message is available.

Returns:

the received message

Throws:

PulsarClientException.AlreadyClosedException - if the consumer was already closed

PulsarClientException.InvalidConfigurationException - if a message listener was defined in the configuration

PulsarClientException

receiveAsync

CompletableFuture<Message<T>> receiveAsync()

Receive a single message

Retrieves a message when it will be available and completes CompletableFuture with received message.

receiveAsync() should be called subsequently once returned CompletableFuture gets complete with received message. Else it creates backlog of receive requests in the application.

Returns:

CompletableFuture<Message> will be completed when message is available

receive

Message<T> receive(int timeout,
                   TimeUnit unit)
            throws PulsarClientException

Receive a single message

Retrieves a message, waiting up to the specified wait time if necessary.

Parameters:

timeout - 0 or less means immediate rather than infinite

unit -

Returns:

the received Message or null if no message available before timeout

Throws:

PulsarClientException.AlreadyClosedException - if the consumer was already closed

PulsarClientException.InvalidConfigurationException - if a message listener was defined in the configuration

PulsarClientException

acknowledge

void acknowledge(Message<?> message)
          throws PulsarClientException

Acknowledge the consumption of a single message

Parameters:

message - The Message to be acknowledged

Throws:

PulsarClientException.AlreadyClosedException - if the consumer was already closed

PulsarClientException

acknowledge

void acknowledge(MessageId messageId)
          throws PulsarClientException

Acknowledge the consumption of a single message, identified by its MessageId

Parameters:

messageId - The MessageId to be acknowledged

Throws:

PulsarClientException.AlreadyClosedException - if the consumer was already closed

PulsarClientException

acknowledgeCumulative

void acknowledgeCumulative(Message<?> message)
                    throws PulsarClientException

Acknowledge the reception of all the messages in the stream up to (and including) the provided message. This method will block until the acknowledge has been sent to the broker. After that, the messages will not be re-delivered to this consumer. Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared. It's equivalent to calling asyncAcknowledgeCumulative(Message) and waiting for the callback to be triggered.

Parameters:

message - The Message to be cumulatively acknowledged

Throws:

PulsarClientException.AlreadyClosedException - if the consumer was already closed

PulsarClientException

acknowledgeCumulative

void acknowledgeCumulative(MessageId messageId)
                    throws PulsarClientException

Acknowledge the reception of all the messages in the stream up to (and including) the provided message. This method will block until the acknowledge has been sent to the broker. After that, the messages will not be re-delivered to this consumer. Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared. It's equivalent to calling asyncAcknowledgeCumulative(MessageId) and waiting for the callback to be triggered.

Parameters:

messageId - The MessageId to be cumulatively acknowledged

Throws:

PulsarClientException.AlreadyClosedException - if the consumer was already closed

PulsarClientException

acknowledgeAsync

CompletableFuture<Void> acknowledgeAsync(Message<?> message)

Asynchronously acknowledge the consumption of a single message

Parameters:

message - The Message to be acknowledged

Returns:

a future that can be used to track the completion of the operation

acknowledgeAsync

CompletableFuture<Void> acknowledgeAsync(MessageId messageId)

Asynchronously acknowledge the consumption of a single message

Parameters:

messageId - The MessageId to be acknowledged

Returns:

a future that can be used to track the completion of the operation

acknowledgeCumulativeAsync

CompletableFuture<Void> acknowledgeCumulativeAsync(Message<?> message)

Asynchronously Acknowledge the reception of all the messages in the stream up to (and including) the provided message. Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared.

Parameters:

message - The Message to be cumulatively acknowledged

Returns:

a future that can be used to track the completion of the operation

acknowledgeCumulativeAsync

CompletableFuture<Void> acknowledgeCumulativeAsync(MessageId messageId)

Asynchronously Acknowledge the reception of all the messages in the stream up to (and including) the provided message. Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared.

Parameters:

messageId - The MessageId to be cumulatively acknowledged

Returns:

a future that can be used to track the completion of the operation

getStats

ConsumerStats getStats()

Get statistics for the consumer.

• numMsgsReceived : Number of messages received in the current interval
• numBytesReceived : Number of bytes received in the current interval
• numReceiveFailed : Number of messages failed to receive in the current interval
• numAcksSent : Number of acks sent in the current interval
• numAcksFailed : Number of acks failed to send in the current interval
• totalMsgsReceived : Total number of messages received
• totalBytesReceived : Total number of bytes received
• totalReceiveFailed : Total number of messages failed to receive
• totalAcksSent : Total number of acks sent
• totalAcksFailed : Total number of acks failed to sent

Returns:

statistic for the consumer

close

void close()
    throws PulsarClientException

Close the consumer and stop the broker to push more messages.

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

Throws:

PulsarClientException

closeAsync

CompletableFuture<Void> closeAsync()

Asynchronously close the consumer and stop the broker to push more messages

Returns:

a future that can be used to track the completion of the operation

hasReachedEndOfTopic

boolean hasReachedEndOfTopic()

Return true if the topic was terminated and this consumer has already consumed all the messages in the topic. Please note that this does not simply mean that the consumer is caught up with the last message published by producers, rather the topic needs to be explicitly "terminated".

redeliverUnacknowledgedMessages

void redeliverUnacknowledgedMessages()

Redelivers all the unacknowledged messages. In Failover mode, the request is ignored if the consumer is not active for the given topic. In Shared mode, the consumers messages to be redelivered are distributed across all the connected consumers. This is a non blocking call and doesn't throw an exception. In case the connection breaks, the messages are redelivered after reconnect.

seek

void seek(MessageId messageId)
   throws PulsarClientException

Reset the subscription associated with this consumer to a specific message id.

The message id can either be a specific message or represent the first or last messages in the topic.

• MessageId.earliest : Reset the subscription on the earliest message available in the topic
• MessageId.latest : Reset the subscription on the latest message in the topic

Note: this operation can only be done on non-partitioned topics. For these, one can rather perform the seek() on the individual partitions.

Parameters:

messageId - the message id where to reposition the subscription

Throws:

PulsarClientException

seekAsync

CompletableFuture<Void> seekAsync(MessageId messageId)

Reset the subscription associated with this consumer to a specific message id.

The message id can either be a specific message or represent the first or last messages in the topic.

• MessageId.earliest : Reset the subscription on the earliest message available in the topic
• MessageId.latest : Reset the subscription on the latest message in the topic

Note: this operation can only be done on non-partitioned topics. For these, one can rather perform the seek() on the individual partitions.

Parameters:

messageId - the message id where to reposition the subscription

Returns:

a future to track the completion of the seek operation

isConnected

boolean isConnected()

Returns:

Whether the consumer is connected to the broker

getConsumerName

String getConsumerName()

Get the name of consumer.

Returns:

consumer name.

pause

void pause()

Stop requesting new messages from the broker until resume() is called. Note that this might cause receive() to block until resume() is called and new messages are pushed by the broker.

resume

void resume()

Resume requesting messages from the broker.