org.springframework.cloud.stream.binder.kafka

Class KafkaMessageChannelBinder

java.lang.Object

org.springframework.cloud.stream.binder.AbstractBinder<org.springframework.messaging.MessageChannel,C,P>

org.springframework.cloud.stream.binder.AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,ExtendedProducerProperties<KafkaProducerProperties>,Collection<org.apache.kafka.common.PartitionInfo>,String>

org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.DisposableBean, org.springframework.beans.factory.InitializingBean, Binder<org.springframework.messaging.MessageChannel,ExtendedConsumerProperties<KafkaConsumerProperties>,ExtendedProducerProperties<KafkaProducerProperties>>, ExtendedBindingProperties<KafkaConsumerProperties,KafkaProducerProperties>, ExtendedPropertiesBinder<org.springframework.messaging.MessageChannel,KafkaConsumerProperties,KafkaProducerProperties>, org.springframework.context.ApplicationContextAware

public class KafkaMessageChannelBinder
extends AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,ExtendedProducerProperties<KafkaProducerProperties>,Collection<org.apache.kafka.common.PartitionInfo>,String>
implements ExtendedPropertiesBinder<org.springframework.messaging.MessageChannel,KafkaConsumerProperties,KafkaProducerProperties>, org.springframework.beans.factory.DisposableBean

A Binder that uses Kafka as the underlying middleware.

Author:

Eric Bottard, Marius Bogoevici, Ilayaperumal Gopinathan, David Turanski, Gary Russell, Mark Fisher, Soby Chacko

Nested Class Summary

Nested classes/interfaces inherited from class org.springframework.cloud.stream.binder.AbstractBinder

AbstractBinder.JavaClassMimeTypeConversion

Field Summary

Fields inherited from class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder

EXPRESSION_PARSER

Fields inherited from class org.springframework.cloud.stream.binder.AbstractBinder

logger

Constructor Summary

Constructors

Constructor and Description
KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties)

Method Summary

Modifier and Type	Method and Description
protected Collection<org.apache.kafka.common.PartitionInfo>	createConsumerDestinationIfNecessary(String name, String group, ExtendedConsumerProperties<KafkaConsumerProperties> properties) Creates the middleware destination the consumer will start to consume data from.
protected org.springframework.integration.core.MessageProducer	createConsumerEndpoint(String name, String group, Collection<org.apache.kafka.common.PartitionInfo> destination, ExtendedConsumerProperties<KafkaConsumerProperties> properties) Creates MessageProducer that receives data from the consumer destination.
protected String	createProducerDestinationIfNecessary(String name, ExtendedProducerProperties<KafkaProducerProperties> properties) Creates target destinations for outbound channels.
protected org.springframework.messaging.MessageHandler	createProducerMessageHandler(String destination, ExtendedProducerProperties<KafkaProducerProperties> producerProperties) Creates a MessageHandler with the ability to send data to the target middleware.
void	destroy()
KafkaConsumerProperties	getExtendedConsumerProperties(String channelName)
KafkaProducerProperties	getExtendedProducerProperties(String channelName)
void	onInit() Subclasses may implement this method to perform any necessary initialization.
void	setAdminUtilsOperation(AdminUtilsOperation adminUtilsOperation)
void	setExtendedBindingProperties(KafkaExtendedBindingProperties extendedBindingProperties)
void	setMetadataRetryOperations(org.springframework.retry.RetryOperations metadataRetryOperations) Retry configuration for operations such as validating topic creation
void	setProducerListener(org.springframework.kafka.support.ProducerListener<byte[],byte[]> producerListener)

Methods inherited from class org.springframework.cloud.stream.binder.AbstractMessageChannelBinder

afterUnbindConsumer, afterUnbindProducer, doBindConsumer, doBindProducer

Methods inherited from class org.springframework.cloud.stream.binder.AbstractBinder

afterPropertiesSet, applyPrefix, bindConsumer, bindProducer, buildPartitionRoutingExpression, buildRetryTemplate, constructDLQName, getApplicationContext, getBeanFactory, groupedName, setApplicationContext, setCodec, setIntegrationEvaluationContext

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.cloud.stream.binder.Binder

bindConsumer, bindProducer

Constructor Detail

KafkaMessageChannelBinder

public KafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties)

Method Detail

setAdminUtilsOperation

public void setAdminUtilsOperation(AdminUtilsOperation adminUtilsOperation)

setMetadataRetryOperations

public void setMetadataRetryOperations(org.springframework.retry.RetryOperations metadataRetryOperations)

Retry configuration for operations such as validating topic creation

Parameters:

metadataRetryOperations - the retry configuration

setExtendedBindingProperties

public void setExtendedBindingProperties(KafkaExtendedBindingProperties extendedBindingProperties)

onInit

public void onInit()
            throws Exception

Description copied from class: AbstractBinder

Subclasses may implement this method to perform any necessary initialization. It will be invoked from AbstractBinder.afterPropertiesSet() which is itself final.

Overrides:

onInit in class AbstractBinder<org.springframework.messaging.MessageChannel,ExtendedConsumerProperties<KafkaConsumerProperties>,ExtendedProducerProperties<KafkaProducerProperties>>

Throws:

Exception

destroy

public void destroy()
             throws Exception

Specified by:

destroy in interface org.springframework.beans.factory.DisposableBean

Throws:

Exception

setProducerListener

public void setProducerListener(org.springframework.kafka.support.ProducerListener<byte[],byte[]> producerListener)

getExtendedConsumerProperties

public KafkaConsumerProperties getExtendedConsumerProperties(String channelName)

Specified by:

getExtendedConsumerProperties in interface ExtendedBindingProperties<KafkaConsumerProperties,KafkaProducerProperties>

getExtendedProducerProperties

public KafkaProducerProperties getExtendedProducerProperties(String channelName)

Specified by:

getExtendedProducerProperties in interface ExtendedBindingProperties<KafkaConsumerProperties,KafkaProducerProperties>

createProducerMessageHandler

protected org.springframework.messaging.MessageHandler createProducerMessageHandler(String destination,
                                                                                    ExtendedProducerProperties<KafkaProducerProperties> producerProperties)
                                                                             throws Exception

Description copied from class: AbstractMessageChannelBinder

Creates a MessageHandler with the ability to send data to the target middleware. If the returned instance is also a Lifecycle, it will be stopped automatically by the binder.

In order to be fully compliant, the MessageHandler of the binder must observe the following headers:

• BinderHeaders.PARTITION_HEADER - indicates the target partition where the message must be sent

Specified by:

createProducerMessageHandler in class AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,ExtendedProducerProperties<KafkaProducerProperties>,Collection<org.apache.kafka.common.PartitionInfo>,String>

Parameters:

destination - the name of the target destination

producerProperties - the producer properties

Returns:

the message handler for sending data to the target middleware

Throws:

Exception

createProducerDestinationIfNecessary

protected String createProducerDestinationIfNecessary(String name,
                                                      ExtendedProducerProperties<KafkaProducerProperties> properties)

Description copied from class: AbstractMessageChannelBinder

Creates target destinations for outbound channels. The implementation is middleware-specific.

Specified by:

createProducerDestinationIfNecessary in class AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,ExtendedProducerProperties<KafkaProducerProperties>,Collection<org.apache.kafka.common.PartitionInfo>,String>

Parameters:

name - the name of the producer destination

properties - producer properties

createConsumerDestinationIfNecessary

protected Collection<org.apache.kafka.common.PartitionInfo> createConsumerDestinationIfNecessary(String name,
                                                                                                 String group,
                                                                                                 ExtendedConsumerProperties<KafkaConsumerProperties> properties)

Description copied from class: AbstractMessageChannelBinder

Creates the middleware destination the consumer will start to consume data from.

Specified by:

createConsumerDestinationIfNecessary in class AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,ExtendedProducerProperties<KafkaProducerProperties>,Collection<org.apache.kafka.common.PartitionInfo>,String>

Parameters:

name - the name of the destination

group - the consumer group

properties - consumer properties

Returns:

reference to the consumer destination

createConsumerEndpoint

protected org.springframework.integration.core.MessageProducer createConsumerEndpoint(String name,
                                                                                      String group,
                                                                                      Collection<org.apache.kafka.common.PartitionInfo> destination,
                                                                                      ExtendedConsumerProperties<KafkaConsumerProperties> properties)

Description copied from class: AbstractMessageChannelBinder

Creates MessageProducer that receives data from the consumer destination. will be started and stopped by the binder.

Specified by:

createConsumerEndpoint in class AbstractMessageChannelBinder<ExtendedConsumerProperties<KafkaConsumerProperties>,ExtendedProducerProperties<KafkaProducerProperties>,Collection<org.apache.kafka.common.PartitionInfo>,String>

Parameters:

name - the name of the target destination

group - the consumer group

destination - reference to the consumer destination

properties - the consumer properties

Returns:

the consumer endpoint.