io.micronaut.configuration.kafka.processor

Class KafkaConsumerProcessor

java.lang.Object

io.micronaut.configuration.kafka.processor.KafkaConsumerProcessor

All Implemented Interfaces:

ConsumerRegistry, io.micronaut.context.processor.AnnotationProcessor<Topic,io.micronaut.inject.ExecutableMethod<?,?>>, io.micronaut.context.processor.ExecutableMethodProcessor<Topic>, java.lang.AutoCloseable

@Singleton
 @Requires(beans=KafkaDefaultConfiguration.class)
public class KafkaConsumerProcessor
extends java.lang.Object
implements io.micronaut.context.processor.ExecutableMethodProcessor<Topic>, java.lang.AutoCloseable, ConsumerRegistry

A ExecutableMethodProcessor that will process all beans annotated with KafkaListener and create and subscribe the relevant methods as consumers to Kafka topics.

Since:

1.0

Author:

Graeme Rocher

Constructor Summary

Constructors

Constructor and Description

KafkaConsumerProcessor(java.util.concurrent.ExecutorService executorService, io.micronaut.runtime.ApplicationConfiguration applicationConfiguration, io.micronaut.context.BeanContext beanContext, AbstractKafkaConsumerConfiguration defaultConsumerConfiguration, ConsumerRecordBinderRegistry binderRegistry, BatchConsumerRecordsBinderRegistry batchBinderRegistry, SerdeRegistry serdeRegistry, ProducerRegistry producerRegistry, KafkaListenerExceptionHandler exceptionHandler, java.util.concurrent.ExecutorService schedulerService, TransactionalProducerRegistry transactionalProducerRegistry) Creates a new processor using the given ExecutorService to schedule consumers on.

Method Summary

Modifier and Type	Method and Description
void	close()
<K,V> org.apache.kafka.clients.consumer.Consumer<K,V>	getConsumer(java.lang.String id) Returns a managed Consumer.
java.util.Set<org.apache.kafka.common.TopicPartition>	getConsumerAssignment(java.lang.String id) Returns a managed Consumer's assignment info.
java.util.Set<java.lang.String>	getConsumerIds() The IDs of the available consumers.
java.util.Set<java.lang.String>	getConsumerSubscription(java.lang.String id) Returns a managed Consumer's subscriptions.
boolean	isPaused(java.lang.String id) Is the consumer with the given ID paused.
boolean	isPaused(java.lang.String id, java.util.Collection<org.apache.kafka.common.TopicPartition> topicPartitions) Is the consumer with the given ID paused to consume from the given topic partitions.
void	pause(java.lang.String id) Pause the consumer for the given ID.
void	pause(java.lang.String id, java.util.Collection<org.apache.kafka.common.TopicPartition> topicPartitions) Pause the consumer for the given ID to consume from the given topic partitions.
void	process(io.micronaut.inject.BeanDefinition<?> beanDefinition, io.micronaut.inject.ExecutableMethod<?,?> method)
void	resume(java.lang.String id) Resume the consumer for the given ID.
void	resume(java.lang.String id, java.util.Collection<org.apache.kafka.common.TopicPartition> topicPartitions) Resume the consumer for the given ID to consume from the given topic partitions.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

KafkaConsumerProcessor

public KafkaConsumerProcessor(@Named(value="consumer")
                              java.util.concurrent.ExecutorService executorService,
                              io.micronaut.runtime.ApplicationConfiguration applicationConfiguration,
                              io.micronaut.context.BeanContext beanContext,
                              AbstractKafkaConsumerConfiguration defaultConsumerConfiguration,
                              ConsumerRecordBinderRegistry binderRegistry,
                              BatchConsumerRecordsBinderRegistry batchBinderRegistry,
                              SerdeRegistry serdeRegistry,
                              ProducerRegistry producerRegistry,
                              KafkaListenerExceptionHandler exceptionHandler,
                              @Named(value="scheduled")
                              java.util.concurrent.ExecutorService schedulerService,
                              TransactionalProducerRegistry transactionalProducerRegistry)

Creates a new processor using the given ExecutorService to schedule consumers on.

Parameters:

executorService - The executor service

applicationConfiguration - The application configuration

beanContext - The bean context

defaultConsumerConfiguration - The default consumer config

binderRegistry - The ConsumerRecordBinderRegistry

batchBinderRegistry - The BatchConsumerRecordsBinderRegistry

serdeRegistry - The Serde registry

producerRegistry - The ProducerRegistry

exceptionHandler - The exception handler to use

schedulerService - The scheduler service

transactionalProducerRegistry - The transactional producer registry

Method Detail

getConsumer

@NonNull
public <K,V> org.apache.kafka.clients.consumer.Consumer<K,V> getConsumer(@NonNull
                                                                                  java.lang.String id)

Description copied from interface: ConsumerRegistry

Returns a managed Consumer. Note that the consumer should not be interacted with directly from a different thread to the poll loop!

Specified by:

getConsumer in interface ConsumerRegistry

Type Parameters:

K - The key generic type

V - The value generic type

Parameters:

id - The id of the producer.

Returns:

The consumer

getConsumerSubscription

@NonNull
public java.util.Set<java.lang.String> getConsumerSubscription(@NonNull
                                                                        java.lang.String id)

Description copied from interface: ConsumerRegistry

Returns a managed Consumer's subscriptions.

Specified by:

getConsumerSubscription in interface ConsumerRegistry

Parameters:

id - The id of the producer.

Returns:

The consumer subscription

getConsumerAssignment

@NonNull
public java.util.Set<org.apache.kafka.common.TopicPartition> getConsumerAssignment(@NonNull
                                                                                            java.lang.String id)

Description copied from interface: ConsumerRegistry

Returns a managed Consumer's assignment info.

Specified by:

getConsumerAssignment in interface ConsumerRegistry

Parameters:

id - The id of the producer.

Returns:

The consumer assignment

getConsumerIds

@NonNull
public java.util.Set<java.lang.String> getConsumerIds()

Description copied from interface: ConsumerRegistry

The IDs of the available consumers.

Specified by:

getConsumerIds in interface ConsumerRegistry

Returns:

The consumers

isPaused

public boolean isPaused(@NonNull
                        java.lang.String id)

Description copied from interface: ConsumerRegistry

Is the consumer with the given ID paused.

Specified by:

isPaused in interface ConsumerRegistry

Parameters:

id - the consumers id

Returns:

True if it is paused

isPaused

public boolean isPaused(@NonNull
                        java.lang.String id,
                        @NonNull
                        java.util.Collection<org.apache.kafka.common.TopicPartition> topicPartitions)

Description copied from interface: ConsumerRegistry

Is the consumer with the given ID paused to consume from the given topic partitions.

Specified by:

isPaused in interface ConsumerRegistry

Parameters:

id - the consumers id

topicPartitions - The topic partitions to check if paused

Returns:

True if it is paused for the topic partitions

pause

public void pause(@NonNull
                  java.lang.String id)

Description copied from interface: ConsumerRegistry

Pause the consumer for the given ID. Note that this method will request that the consumer is paused, however does not block until the consumer is actually paused. You can use the ConsumerRegistry.isPaused(String) method to establish when the consumer has actually been paused.

Specified by:

pause in interface ConsumerRegistry

Parameters:

id - The id of the consumer

pause

public void pause(@NonNull
                  java.lang.String id,
                  @NonNull
                  java.util.Collection<org.apache.kafka.common.TopicPartition> topicPartitions)

Description copied from interface: ConsumerRegistry

Pause the consumer for the given ID to consume from the given topic partitions. Note that this method will request that the consumer is paused, however does not block until the consumer is actually paused. You can use the ConsumerRegistry.isPaused(String, Collection) method to establish when the consumer has actually been paused for the topic partitions.

Specified by:

pause in interface ConsumerRegistry

Parameters:

id - The id of the consumer

topicPartitions - The topic partitions to pause consuming from

resume

public void resume(@NonNull
                   java.lang.String id)

Description copied from interface: ConsumerRegistry

Resume the consumer for the given ID. Note that this method will request that the consumer is resumed, however does not block until the consumer is actually resumed. You can use the ConsumerRegistry.isPaused(String) method to establish when the consumer has actually been resumed.

Specified by:

resume in interface ConsumerRegistry

Parameters:

id - The id of the consumer

resume

public void resume(@NonNull
                   java.lang.String id,
                   @NonNull
                   java.util.Collection<org.apache.kafka.common.TopicPartition> topicPartitions)

Description copied from interface: ConsumerRegistry

Resume the consumer for the given ID to consume from the given topic partitions. Note that this method will request that the consumer is resumed, however does not block until the consumer is actually resumed. You can use the ConsumerRegistry.isPaused(String, Collection) method to establish when the consumer has actually been resumed to consume from the given topic partitions.

Specified by:

resume in interface ConsumerRegistry

Parameters:

id - The id of the consumer

topicPartitions - The topic partitions to pause consuming from

process

public void process(io.micronaut.inject.BeanDefinition<?> beanDefinition,
                    io.micronaut.inject.ExecutableMethod<?,?> method)

Specified by:

process in interface io.micronaut.context.processor.AnnotationProcessor<Topic,io.micronaut.inject.ExecutableMethod<?,?>>

Specified by:

process in interface io.micronaut.context.processor.ExecutableMethodProcessor<Topic>

close

@PreDestroy
public void close()

Specified by:

close in interface java.lang.AutoCloseable