io.micronaut.pulsar.annotation

Annotation Type PulsarConsumer

@Documented
 @Retention(value=RUNTIME)
 @Target(value=METHOD)
 @Executable
 @MessageListener
public @interface PulsarConsumer

Marks a method as a Pulsar Consumer.

Since:

1.0

Author:

Haris Secic

Required Element Summary

Required Elements

Modifier and Type	Required Element and Description
String	consumerName

Optional Element Summary

Optional Elements

Modifier and Type	Optional Element and Description
String	ackTimeout By default, it will use default value of ConsumerBuilder which is disabled and no redelivery happens unless consumer crashed.
String	deadLetterTopic Default is fallback to Pulsar Client Java library.
org.apache.pulsar.common.schema.KeyValueEncodingType	keyEncoding Ignored if annotated method does not contain parameter annotated with MessageKey.
MessageSchema	keySchema Ignored if keyEncoding() is KeyValueEncodingType.INLINE or if MessageKey is missing on one of the attributes of the annotated method.
int	maxRetriesBeforeDlq
int	patternAutoDiscoveryPeriod Used in combination with topicsPattern().
int	priorityLevel By default no priority is set.
int	receiverQueueSize
MessageSchema	schema Defaults to MessageSchema.BYTES as default value for Pulsar Schema is byte[].
boolean	subscribeAsync By default, consumer should subscribe in non-blocking manner using default CompletableFuture of ConsumerBuilder.subscribeAsync().
String	subscription
org.apache.pulsar.client.api.RegexSubscriptionMode	subscriptionTopicsMode Ignored if topics() attribute is set.
org.apache.pulsar.client.api.SubscriptionType	subscriptionType
@Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w") String	topic Has precedence over topics() and topicsPattern().
@Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w") String[]	topics Has precedence over topicPattern.
@Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-?\\w+)?/).+") String	topicsPattern Ignored if topics attribute is set.
@Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w") String	value

Element Detail

consumerName

public abstract String consumerName

Returns:

Consumer name for more descriptive monitoring

value

@AliasFor(member="topic") @AliasFor(annotation=io.micronaut.messaging.annotation.MessageMapping.class,member="value")
 @Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w")
public abstract @Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w") String value

Returns:

Same as topic()

Default:

""

topic

@Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w")
 @AliasFor(member="value") @AliasFor(annotation=io.micronaut.messaging.annotation.MessageMapping.class,member="value")
public abstract @Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w") String topic

Has precedence over topics() and topicsPattern(). Single topic to consume messages from.

Returns:

Topic name to listen to

Default:

""

topics

@Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w")
 @AliasFor(annotation=io.micronaut.messaging.annotation.MessageMapping.class,
          member="value")
public abstract @Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-|\\w+)*\\w/)\\w+(-|\\w+)*\\w") String[] topics

Has precedence over topicPattern.

Returns:

List of topic names in form of (persistent|non-persistent)://tenant-name/namespace/topic.

Default:

{}

topicsPattern

@Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-?\\w+)?/).+")
 @AliasFor(annotation=io.micronaut.messaging.annotation.MessageMapping.class,
          member="value")
public abstract @Pattern(regexp="((non-)?persistent://)?((\\w+(-|\\w+)*\\w/)|(\\$\\{tenant\\}))(\\w+(-?\\w+)?/).+") String topicsPattern

Ignored if topics attribute is set.

Returns:

Topics name in form of tenantName/namespace/topic-name-pattern.

Default:

""

schema

public abstract MessageSchema schema

Defaults to MessageSchema.BYTES as default value for Pulsar Schema is byte[].

Returns:

Schema to use with pulsar topic consumer

Default:

io.micronaut.pulsar.MessageSchema.BYTES

keySchema

public abstract MessageSchema keySchema

Ignored if keyEncoding() is KeyValueEncodingType.INLINE or if MessageKey is missing on one of the attributes of the annotated method.

Returns:

Schema to use while parsing message key from Pulsar message

Default:

io.micronaut.pulsar.MessageSchema.BYTES

keyEncoding

public abstract org.apache.pulsar.common.schema.KeyValueEncodingType keyEncoding

Ignored if annotated method does not contain parameter annotated with MessageKey. Indicates to message processor is the key inside message payload or stored separately in which case key can have different schema from message body. Type KeyValueEncodingType.INLINE can only work for JSON or Avro schemas; otherwise throws exception while reading.

Returns:

Whether to read key from the message payload or separately.

Default:

org.apache.pulsar.common.schema.KeyValueEncodingType.INLINE

subscription

public abstract String subscription

Returns:

Subscription name in case consumer was defined outside the PulsarSubscription annotated class

Default:

""

subscriptionType

public abstract org.apache.pulsar.client.api.SubscriptionType subscriptionType

Returns:

Subscription type in case consumer was defined outside of PulsarSubscription annotated class

Default:

org.apache.pulsar.client.api.SubscriptionType.Exclusive

subscriptionTopicsMode

public abstract org.apache.pulsar.client.api.RegexSubscriptionMode subscriptionTopicsMode

Ignored if topics() attribute is set. Default value RegexSubscriptionMode.AllTopics

Whether to read topics from persistent, or non-persistent storage, or both.

Returns:

subscription

Default:

org.apache.pulsar.client.api.RegexSubscriptionMode.AllTopics

patternAutoDiscoveryPeriod

public abstract int patternAutoDiscoveryPeriod

Used in combination with topicsPattern(). Ignored using topics(). Must be greater than 1. Low values should be avoided. Pulsar default value is 1 minute

Returns:

Amount of delay between checks, in seconds, for new topic matching given pattern.

Default:

-1

subscribeAsync

public abstract boolean subscribeAsync

By default, consumer should subscribe in non-blocking manner using default CompletableFuture of ConsumerBuilder.subscribeAsync().

If blocking set to false application will block until consumer is successfully subscribed. This is different from the actual process of consuming the messages which still happens in separate thread managed by the underlying Pulsar Client library.

Returns:

Should the consumer subscribe in async manner or blocking

Default:

true

ackTimeout

public abstract String ackTimeout

By default, it will use default value of ConsumerBuilder which is disabled and no redelivery happens unless consumer crashed.

Must be greater than 1s.

Returns:

Allowed time to pass before message is acknowledged.

See Also:

ConsumerBuilder.ackTimeout(long, java.util.concurrent.TimeUnit)

Default:

""

receiverQueueSize

public abstract int receiverQueueSize

Returns:

Number of items allowed in the queue. Default 1000 as in Pulsar Java Client

Default:

1000

priorityLevel

public abstract int priorityLevel

By default no priority is set. Use any value less than 0 to disable. Use anything above 0 to set lower priority level.

Returns:

priority level for a consumer

See Also:

ConsumerBuilder.priorityLevel(int)

Default:

-1

deadLetterTopic

public abstract String deadLetterTopic

Default is fallback to Pulsar Client Java library.

Returns:

Dead Letter topic name.

Default:

""

maxRetriesBeforeDlq

public abstract int maxRetriesBeforeDlq

Returns:

Maximum numbers of retires before sending message to dead letter queue topic.

Default:

3