Package org.flowable.eventregistry.spring.kafka

Class KafkaChannelDefinitionProcessor

java.lang.Object

org.flowable.eventregistry.spring.kafka.KafkaChannelDefinitionProcessor

All Implemented Interfaces:

EventListener, ChannelModelProcessor, Aware, BeanFactoryAware, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent>

public class KafkaChannelDefinitionProcessor
extends Object
implements BeanFactoryAware, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent>, ChannelModelProcessor

A ChannelModelProcessor which is responsible for configuring Kafka Event registry integration. This class is not meant to be extended.

Author:

Filip Hrisafov

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected static class	KafkaChannelDefinitionProcessor.Configuration
protected static class	KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration
protected static class	KafkaChannelDefinitionProcessor.RetryTopicContainerFactoryDecorator

Field Summary

Fields

Modifier and Type	Field	Description
protected ApplicationContext	applicationContext
protected BeanFactory	beanFactory
static final String	CHANNEL_ID_PREFIX
protected KafkaListenerContainerFactory<?>	containerFactory
protected String	containerFactoryBeanName
protected boolean	contextRefreshed
protected static final int	DEFAULT_PARTITION_FOR_MANUAL_ASSIGNMENT
protected StringValueResolver	embeddedValueResolver
protected KafkaListenerEndpointRegistry	endpointRegistry
protected BeanExpressionContext	expressionContext
protected KafkaAdminOperations	kafkaAdminOperations
protected KafkaConsumerBackoffManager	kafkaConsumerBackoffManager
protected KafkaOperations<Object,Object>	kafkaOperations
protected final org.slf4j.Logger	logger
protected com.fasterxml.jackson.databind.ObjectMapper	objectMapper
protected BeanExpressionResolver	resolver
protected Map<String,Collection<String>>	retryEndpointsByMainEndpointId

Constructor Summary

Constructors

Constructor	Description
KafkaChannelDefinitionProcessor(com.fasterxml.jackson.databind.ObjectMapper objectMapper)

Method Summary

Modifier and Type	Method	Description
boolean	canProcess(ChannelModel channelModel)
boolean	canProcessIfChannelModelAlreadyRegistered(ChannelModel channelModel)
protected Collection<KafkaChannelDefinitionProcessor.Configuration>	createEndpointConfigurations(KafkaInboundChannelModel channelModel, String tenantId, EventRegistry eventRegistry, KafkaListenerEndpoint mainEndpoint, KafkaListenerContainerFactory<?> containerFactory)
protected KafkaListenerEndpoint	createKafkaListenerEndpoint(KafkaInboundChannelModel channelModel, String tenantId, EventRegistry eventRegistry)
protected ListenerContainerFactoryConfigurer	createListenerContainerFactoryConfigurer(KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration retryConfiguration, BackOff backOff, DefaultDestinationTopicResolver topicResolver)
protected GenericMessageListener<org.apache.kafka.clients.consumer.ConsumerRecord<Object,Object>>	createMessageListener(EventRegistry eventRegistry, InboundChannelModel inboundChannelModel)
protected void	createNewTopics(Collection<String> topics, int numPartitions, short replicationFactor)
protected org.springframework.retry.backoff.SleepingBackOffPolicy<?>	createNonBlockingBackOffPolicy(KafkaInboundChannelModel.NonBlockingRetryBackOff backOff)
protected RetryTopicConfiguration	createRetryTopicConfiguration(KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration retryConfiguration)
protected KafkaListenerContainerFactory<?>	decorateFactory(DestinationTopic.Properties destinationTopicProperties, ListenerContainerFactoryConfigurer factoryConfigurer, RetryTopicConfiguration retryTopicConfiguration)
KafkaListenerContainerFactory<?>	getContainerFactory()
String	getContainerFactoryBeanName()
protected String	getEndpointGroupId(KafkaInboundChannelModel channelDefinition, String id)
protected String	getEndpointId(ChannelModel channelModel, String tenantId)
KafkaListenerEndpointRegistry	getEndpointRegistry()
KafkaAdminOperations	getKafkaAdminOperations()
KafkaConsumerBackoffManager	getKafkaConsumerBackoffManager()
KafkaOperations<Object,Object>	getKafkaOperations()
protected KafkaConsumerBackoffManager	getOrCreateKafkaConsumerBackoffManager()
protected TaskScheduler	getOrCreateRetryTopicTaskScheduler()
protected Consumer<Collection<String>>	getTopicCreationFunction(KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration retryConfiguration)
protected static Collection<TopicPartitionOffset>	getTopicPartitions(DestinationTopic.Properties properties, Suffixer suffixer, TopicPartitionOffset[] topicPartitionOffsets)
protected static TopicPartitionOffset	getTPOForMainTopic(Suffixer suffixer, TopicPartitionOffset tpo)
protected static TopicPartitionOffset	getTPOForRetryTopics(DestinationTopic.Properties properties, Suffixer suffixer, TopicPartitionOffset tpo)
void	onApplicationEvent(ContextRefreshedEvent event)
protected Stream<Integer>	parsePartitions(String partsString)	Parse a list of partitions into a List.
protected void	processAndRegisterEndpoints(KafkaInboundChannelModel channelModel, String tenantId, EventRegistry eventRegistry)
protected void	processOutboundDefinition(KafkaOutboundChannelModel channelModel)
void	registerChannelModel(ChannelModel channelModel, String tenantId, EventRegistry eventRegistry, EventRepositoryService eventRepositoryService, boolean fallbackToDefaultTenant)
protected void	registerEndpoint(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory)	Register a new KafkaListenerEndpoint alongside the KafkaListenerContainerFactory to use to create the underlying container.
protected String	resolve(String value)
protected KafkaListenerContainerFactory<?>	resolveContainerFactory(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> containerFactory)
protected Object	resolveExpression(String value)
protected <T> T	resolveExpression(String expression, Class<T> type)
protected Boolean	resolveExpressionAsBoolean(String value, String attribute)
protected Boolean	resolveExpressionAsBoolean(String value, String attribute, Boolean defaultValue)
protected Double	resolveExpressionAsDouble(String value, String attribute)
protected Integer	resolveExpressionAsInteger(String value, String attribute)
protected Integer	resolveExpressionAsInteger(String value, String attribute, Integer defaultValue)
protected Long	resolveExpressionAsLong(String value, String attribute)
protected String	resolveExpressionAsString(String value, String attribute)
protected KafkaMessageKeyProvider<?>	resolveKafkaMessageKeyProvider(KafkaOutboundChannelModel channelModel)
protected KafkaPartitionProvider	resolveKafkaPartitionProvider(KafkaOutboundChannelModel channelModel)
protected void	resolvePartitionAsInteger(String topic, Object resolvedValue, List<TopicPartitionOffset> result)
protected Pattern	resolvePattern(KafkaInboundChannelModel channelModel)
protected Properties	resolveProperties(List<KafkaInboundChannelModel.CustomProperty> consumerProperties)
protected KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration	resolveRetryConfiguration(KafkaInboundChannelModel channelModel)
protected Collection<TopicPartitionOffset>	resolveTopicPartitions(KafkaInboundChannelModel channelModel)
protected void	resolveTopics(Object resolvedValue, List<String> result, KafkaInboundChannelModel channelDefinition)
protected Collection<String>	resolveTopics(KafkaInboundChannelModel channelDefinition)
void	setApplicationContext(ApplicationContext applicationContext)
void	setBeanFactory(BeanFactory beanFactory)
void	setContainerFactory(KafkaListenerContainerFactory<?> containerFactory)
void	setContainerFactoryBeanName(String containerFactoryBeanName)
void	setEndpointRegistry(KafkaListenerEndpointRegistry endpointRegistry)
void	setKafkaAdminOperations(KafkaAdminOperations kafkaAdminOperations)
void	setKafkaConsumerBackoffManager(KafkaConsumerBackoffManager kafkaConsumerBackoffManager)
void	setKafkaOperations(KafkaOperations<Object,Object> kafkaOperations)
void	unregisterChannelModel(ChannelModel channelModel, String tenantId, EventRepositoryService eventRepositoryService)
protected void	unregisterEndpoint(String endpointId, ChannelModel channelModel, String tenantId)

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.context.ApplicationListener

supportsAsyncExecution

Field Details

CHANNEL_ID_PREFIX

public static final String CHANNEL_ID_PREFIX

See Also:

• Constant Field Values

DEFAULT_PARTITION_FOR_MANUAL_ASSIGNMENT

protected static final int DEFAULT_PARTITION_FOR_MANUAL_ASSIGNMENT

See Also:

• Constant Field Values

logger

protected final org.slf4j.Logger logger

kafkaOperations

protected KafkaOperations<Object,Object> kafkaOperations

kafkaAdminOperations

protected KafkaAdminOperations kafkaAdminOperations

endpointRegistry

protected KafkaListenerEndpointRegistry endpointRegistry

containerFactoryBeanName

protected String containerFactoryBeanName

containerFactory

protected KafkaListenerContainerFactory<?> containerFactory

kafkaConsumerBackoffManager

protected KafkaConsumerBackoffManager kafkaConsumerBackoffManager

beanFactory

protected BeanFactory beanFactory

applicationContext

protected ApplicationContext applicationContext

contextRefreshed

protected boolean contextRefreshed

objectMapper

protected com.fasterxml.jackson.databind.ObjectMapper objectMapper

resolver

protected BeanExpressionResolver resolver

embeddedValueResolver

protected StringValueResolver embeddedValueResolver

expressionContext

protected BeanExpressionContext expressionContext

retryEndpointsByMainEndpointId

protected Map<String,Collection<String>> retryEndpointsByMainEndpointId

Constructor Details

KafkaChannelDefinitionProcessor

public KafkaChannelDefinitionProcessor(com.fasterxml.jackson.databind.ObjectMapper objectMapper)

Method Details

canProcess

public boolean canProcess(ChannelModel channelModel)

Specified by:

canProcess in interface ChannelModelProcessor

canProcessIfChannelModelAlreadyRegistered

public boolean canProcessIfChannelModelAlreadyRegistered(ChannelModel channelModel)

Specified by:

canProcessIfChannelModelAlreadyRegistered in interface ChannelModelProcessor

registerChannelModel

public void registerChannelModel(ChannelModel channelModel,
 String tenantId,
 EventRegistry eventRegistry,
 EventRepositoryService eventRepositoryService,
 boolean fallbackToDefaultTenant)

Specified by:

registerChannelModel in interface ChannelModelProcessor

createKafkaListenerEndpoint

protected KafkaListenerEndpoint createKafkaListenerEndpoint(KafkaInboundChannelModel channelModel,
 String tenantId,
 EventRegistry eventRegistry)

processAndRegisterEndpoints

protected void processAndRegisterEndpoints(KafkaInboundChannelModel channelModel,
 String tenantId,
 EventRegistry eventRegistry)

createEndpointConfigurations

protected Collection<KafkaChannelDefinitionProcessor.Configuration> createEndpointConfigurations(KafkaInboundChannelModel channelModel,
 String tenantId,
 EventRegistry eventRegistry,
 KafkaListenerEndpoint mainEndpoint,
 KafkaListenerContainerFactory<?> containerFactory)

getTopicPartitions

protected static Collection<TopicPartitionOffset> getTopicPartitions(DestinationTopic.Properties properties,
 Suffixer suffixer,
 TopicPartitionOffset[] topicPartitionOffsets)

getTPOForRetryTopics

protected static TopicPartitionOffset getTPOForRetryTopics(DestinationTopic.Properties properties,
 Suffixer suffixer,
 TopicPartitionOffset tpo)

getTPOForMainTopic

protected static TopicPartitionOffset getTPOForMainTopic(Suffixer suffixer,
 TopicPartitionOffset tpo)

getTopicCreationFunction

protected Consumer<Collection<String>> getTopicCreationFunction(KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration retryConfiguration)

createNewTopics

protected void createNewTopics(Collection<String> topics,
 int numPartitions,
 short replicationFactor)

createListenerContainerFactoryConfigurer

protected ListenerContainerFactoryConfigurer createListenerContainerFactoryConfigurer(KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration retryConfiguration,
 BackOff backOff,
 DefaultDestinationTopicResolver topicResolver)

getOrCreateKafkaConsumerBackoffManager

protected KafkaConsumerBackoffManager getOrCreateKafkaConsumerBackoffManager()

getOrCreateRetryTopicTaskScheduler

protected TaskScheduler getOrCreateRetryTopicTaskScheduler()

decorateFactory

protected KafkaListenerContainerFactory<?> decorateFactory(DestinationTopic.Properties destinationTopicProperties,
 ListenerContainerFactoryConfigurer factoryConfigurer,
 RetryTopicConfiguration retryTopicConfiguration)

processOutboundDefinition

protected void processOutboundDefinition(KafkaOutboundChannelModel channelModel)

resolveExpressionAsInteger

protected Integer resolveExpressionAsInteger(String value,
 String attribute)

resolveExpressionAsInteger

protected Integer resolveExpressionAsInteger(String value,
 String attribute,
 Integer defaultValue)

resolveExpressionAsLong

protected Long resolveExpressionAsLong(String value,
 String attribute)

resolveExpressionAsDouble

protected Double resolveExpressionAsDouble(String value,
 String attribute)

resolveExpressionAsBoolean

protected Boolean resolveExpressionAsBoolean(String value,
 String attribute)

resolveExpressionAsBoolean

protected Boolean resolveExpressionAsBoolean(String value,
 String attribute,
 Boolean defaultValue)

resolveExpressionAsString

protected String resolveExpressionAsString(String value,
 String attribute)

resolveTopics

protected Collection<String> resolveTopics(KafkaInboundChannelModel channelDefinition)

resolveTopics

protected void resolveTopics(Object resolvedValue,
 List<String> result,
 KafkaInboundChannelModel channelDefinition)

resolvePattern

protected Pattern resolvePattern(KafkaInboundChannelModel channelModel)

resolveTopicPartitions

protected Collection<TopicPartitionOffset> resolveTopicPartitions(KafkaInboundChannelModel channelModel)

resolvePartitionAsInteger

protected void resolvePartitionAsInteger(String topic,
 Object resolvedValue,
 List<TopicPartitionOffset> result)

parsePartitions

protected Stream<Integer> parsePartitions(String partsString)

Parse a list of partitions into a List. Example: "0-5,10-15". This parsing is the same as it is done in the Spring KafkaListenerAnnotationBeanPostProcessor.

Parameters:

partsString - the comma-delimited list of partitions/ranges.

Returns:

the stream of partition numbers, sorted and de-duplicated.

resolveKafkaPartitionProvider

protected KafkaPartitionProvider resolveKafkaPartitionProvider(KafkaOutboundChannelModel channelModel)

resolveKafkaMessageKeyProvider

protected KafkaMessageKeyProvider<?> resolveKafkaMessageKeyProvider(KafkaOutboundChannelModel channelModel)

resolveExpression

protected <T> T resolveExpression(String expression,
 Class<T> type)

resolveExpression

protected Object resolveExpression(String value)

createMessageListener

protected GenericMessageListener<org.apache.kafka.clients.consumer.ConsumerRecord<Object,Object>> createMessageListener(EventRegistry eventRegistry,
 InboundChannelModel inboundChannelModel)

unregisterChannelModel

public void unregisterChannelModel(ChannelModel channelModel,
 String tenantId,
 EventRepositoryService eventRepositoryService)

Specified by:

unregisterChannelModel in interface ChannelModelProcessor

unregisterEndpoint

protected void unregisterEndpoint(String endpointId,
 ChannelModel channelModel,
 String tenantId)

registerEndpoint

protected void registerEndpoint(KafkaListenerEndpoint endpoint,
 KafkaListenerContainerFactory<?> factory)

Register a new KafkaListenerEndpoint alongside the KafkaListenerContainerFactory to use to create the underlying container.

The factory may be null if the default factory has to be used for that endpoint.

resolveContainerFactory

protected KafkaListenerContainerFactory<?> resolveContainerFactory(KafkaListenerEndpoint endpoint,
 KafkaListenerContainerFactory<?> containerFactory)

getEndpointId

protected String getEndpointId(ChannelModel channelModel,
 String tenantId)

getEndpointGroupId

protected String getEndpointGroupId(KafkaInboundChannelModel channelDefinition,
 String id)

resolve

protected String resolve(String value)

resolveProperties

protected Properties resolveProperties(List<KafkaInboundChannelModel.CustomProperty> consumerProperties)

setBeanFactory

public void setBeanFactory(BeanFactory beanFactory)
                    throws BeansException

Specified by:

setBeanFactory in interface BeanFactoryAware

Throws:

BeansException

setApplicationContext

public void setApplicationContext(ApplicationContext applicationContext)
                           throws BeansException

Specified by:

setApplicationContext in interface ApplicationContextAware

Throws:

BeansException

onApplicationEvent

public void onApplicationEvent(ContextRefreshedEvent event)

Specified by:

onApplicationEvent in interface ApplicationListener<ContextRefreshedEvent>

getKafkaOperations

public KafkaOperations<Object,Object> getKafkaOperations()

setKafkaOperations

public void setKafkaOperations(KafkaOperations<Object,Object> kafkaOperations)

getKafkaAdminOperations

public KafkaAdminOperations getKafkaAdminOperations()

setKafkaAdminOperations

public void setKafkaAdminOperations(KafkaAdminOperations kafkaAdminOperations)

getEndpointRegistry

public KafkaListenerEndpointRegistry getEndpointRegistry()

setEndpointRegistry

public void setEndpointRegistry(KafkaListenerEndpointRegistry endpointRegistry)

getContainerFactoryBeanName

public String getContainerFactoryBeanName()

setContainerFactoryBeanName

public void setContainerFactoryBeanName(String containerFactoryBeanName)

getContainerFactory

public KafkaListenerContainerFactory<?> getContainerFactory()

setContainerFactory

public void setContainerFactory(KafkaListenerContainerFactory<?> containerFactory)

getKafkaConsumerBackoffManager

public KafkaConsumerBackoffManager getKafkaConsumerBackoffManager()

setKafkaConsumerBackoffManager

public void setKafkaConsumerBackoffManager(KafkaConsumerBackoffManager kafkaConsumerBackoffManager)

createRetryTopicConfiguration

protected RetryTopicConfiguration createRetryTopicConfiguration(KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration retryConfiguration)

resolveRetryConfiguration

protected KafkaChannelDefinitionProcessor.ResolvedRetryConfiguration resolveRetryConfiguration(KafkaInboundChannelModel channelModel)

createNonBlockingBackOffPolicy

protected org.springframework.retry.backoff.SleepingBackOffPolicy<?> createNonBlockingBackOffPolicy(KafkaInboundChannelModel.NonBlockingRetryBackOff backOff)