io.micronaut.configuration.kafka.annotation

Annotation Type KafkaListener

@Documented
 @Retention(value=RUNTIME)
 @Target(value={TYPE,METHOD,ANNOTATION_TYPE})
 @MessageListener
public @interface KafkaListener

Annotation applied at the class level to indicate that a bean is a Kafka Consumer.

Since:

1.0

Author:

graemerocher

Optional Element Summary

Optional Elements

Modifier and Type	Optional Element and Description
boolean	autoStartup Setting to allow start the consumer in paused mode.
boolean	batch By default each listener will consume a single ConsumerRecord.
java.lang.String	clientId Sets the client id of the Kafka consumer.
ErrorStrategy	errorStrategy Setting the error strategy allows you to resume at the next offset or to seek the consumer (stop on error) to the failed offset so that it can retry if an error occurs The consumer bean is still able to implement a custom exception handler to replace DefaultKafkaListenerExceptionHandler and set the error strategy.
java.lang.String	groupId The same as value().
java.lang.String	heartbeatInterval The heart beat interval for the consumer.
org.apache.kafka.common.IsolationLevel	isolation Kafka consumer isolation level to control how to read messages written transactionally.
OffsetReset	offsetReset
OffsetStrategy	offsetStrategy The OffsetStrategy to use for the consumer.
java.lang.String	pollTimeout The timeout to use for calls to Consumer.poll(java.time.Duration).
java.lang.String	producerClientId The client id of the producer that is used for SendTo.
java.lang.String	producerTransactionalId This setting applies only for the producer that is used for SendTo.
io.micronaut.context.annotation.Property[]	properties Additional properties to configure with for Consumer.
boolean	redelivery For listeners that return reactive types, message offsets are committed without blocking the consumer.
java.lang.String	sessionTimeout The session timeout for a consumer.
int	threads Kafka consumers are by default single threaded.
boolean	uniqueGroupId A unique string (UUID) can be appended to the group ID.
java.lang.String	value Sets the consumer group id of the Kafka consumer.

Element Detail

value

@AliasFor(member="groupId")
public abstract java.lang.String value

Sets the consumer group id of the Kafka consumer. If not specified the group id is configured to be the value of ApplicationConfiguration.getName() otherwise the name of the class is used.

Returns:

The group id

Default:

""

groupId

@AliasFor(member="value")
public abstract java.lang.String groupId

The same as value().

Returns:

The group id

Default:

""

uniqueGroupId

public abstract boolean uniqueGroupId

A unique string (UUID) can be appended to the group ID. In that case, each consumer will be the only member of a unique consumer group.

Returns:

True to make each group ID unique. Defaults to false.

Default:

false

clientId

public abstract java.lang.String clientId

Sets the client id of the Kafka consumer. If not specified the client id is configured to be the value of ApplicationConfiguration.getName().

Returns:

The client id

Default:

""

offsetStrategy

public abstract OffsetStrategy offsetStrategy

The OffsetStrategy to use for the consumer.

Returns:

The OffsetStrategy

Default:

io.micronaut.configuration.kafka.annotation.OffsetStrategy.AUTO

offsetReset

public abstract OffsetReset offsetReset

Returns:

The strategy to use to start consuming records

Default:

io.micronaut.configuration.kafka.annotation.OffsetReset.LATEST

producerClientId

public abstract java.lang.String producerClientId

The client id of the producer that is used for SendTo.

Returns:

the producer client id

Default:

""

producerTransactionalId

public abstract java.lang.String producerTransactionalId

This setting applies only for the producer that is used for SendTo. The TransactionalId to use for transactional delivery. This enables reliability semantics which span multiple producer sessions since it allows the client to guarantee that transactions using the same TransactionalId have been completed prior to starting any new transactions. If no TransactionalId is provided, then the producer is limited to idempotent delivery. If a TransactionalId is configured, enable.idempotence is implied. By default, the TransactionId is not configured, which means transactions cannot be used.

Returns:

the producer transaction id

Default:

""

isolation

public abstract org.apache.kafka.common.IsolationLevel isolation

Kafka consumer isolation level to control how to read messages written transactionally. See ConsumerConfig.ISOLATION_LEVEL_CONFIG.

Returns:

The isolation level

Default:

org.apache.kafka.common.IsolationLevel.READ_UNCOMMITTED

errorStrategy

public abstract ErrorStrategy errorStrategy

Setting the error strategy allows you to resume at the next offset or to seek the consumer (stop on error) to the failed offset so that it can retry if an error occurs The consumer bean is still able to implement a custom exception handler to replace DefaultKafkaListenerExceptionHandler and set the error strategy.

Returns:

The strategy to use when an error occurs

Default:

@io.micronaut.configuration.kafka.annotation.ErrorStrategy

threads

public abstract int threads

Kafka consumers are by default single threaded. If you wish to increase the number of threads for a consumer you can alter this setting. Note that this means that multiple partitions will be allocated to a single application.

NOTE: When using this setting if your bean is Singleton then local state will be s shared between invocations from different consumer threads

Returns:

The number of threads

Default:

1

pollTimeout

public abstract java.lang.String pollTimeout

The timeout to use for calls to Consumer.poll(java.time.Duration).

Returns:

The timeout. Defaults to 100ms

Default:

"100ms"

sessionTimeout

public abstract java.lang.String sessionTimeout

The session timeout for a consumer. See session.timeout.ms.

Returns:

The session timeout as a duration.

See Also:

ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG

Default:

""

heartbeatInterval

public abstract java.lang.String heartbeatInterval

The heart beat interval for the consumer. See heartbeat.interval.ms.

Returns:

The heartbeat interval as a duration.

See Also:

ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG

Default:

""

redelivery

public abstract boolean redelivery

For listeners that return reactive types, message offsets are committed without blocking the consumer. If the reactive type produces an error, then the message can be lost. Enabling this setting allows the consumer to redelivery the message so that it can be processed again by another consumer that may have better luck.

Returns:

True if redelivery should be enabled. Defaults to false.

Default:

false

batch

public abstract boolean batch

By default each listener will consume a single ConsumerRecord. By setting this value to true and specifying a container type in the method signatures you can indicate that the method should instead receive all the records at once in a batch.

Returns:

Whether to receive a batch of records or not

Default:

false

properties

public abstract io.micronaut.context.annotation.Property[] properties

Additional properties to configure with for Consumer.

Returns:

The properties

Default:

{}

autoStartup

public abstract boolean autoStartup

Setting to allow start the consumer in paused mode.

Returns:

the auto startup setting

Default:

true