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<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

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

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.InitializingBean, Binder<org.springframework.messaging.MessageChannel,ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>>, ExtendedBindingProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties,org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>, ExtendedPropertiesBinder<org.springframework.messaging.MessageChannel,org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties,org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>, PollableConsumerBinder<org.springframework.messaging.MessageHandler,ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>>, org.springframework.context.ApplicationContextAware, org.springframework.context.ApplicationEventPublisherAware

public class KafkaMessageChannelBinder
extends AbstractMessageChannelBinder<ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>
implements ExtendedPropertiesBinder<org.springframework.messaging.MessageChannel,org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties,org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>

A Binder that uses Kafka as the underlying middleware.

Author:

Eric Bottard, Marius Bogoevici, Ilayaperumal Gopinathan, David Turanski, Gary Russell, Mark Fisher, Soby Chacko, Henryk Konsek, Doug Saus

Nested Class Summary

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

AbstractMessageChannelBinder.ErrorInfrastructure, AbstractMessageChannelBinder.PolledConsumerResources

Field Summary

Fields

Modifier and Type	Field and Description
static String	X_EXCEPTION_FQCN
static String	X_EXCEPTION_MESSAGE
static String	X_EXCEPTION_STACKTRACE
static String	X_ORIGINAL_OFFSET
static String	X_ORIGINAL_PARTITION
static String	X_ORIGINAL_TIMESTAMP
static String	X_ORIGINAL_TIMESTAMP_TYPE
static String	X_ORIGINAL_TOPIC

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

provisioningProvider

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

logger

Constructor Summary

Constructors

Constructor and Description
KafkaMessageChannelBinder(org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties configurationProperties, org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner provisioningProvider)
KafkaMessageChannelBinder(org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties configurationProperties, org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner provisioningProvider, ListenerContainerCustomizer<org.springframework.kafka.listener.AbstractMessageListenerContainer<?,?>> containerCustomizer)

Method Summary

Modifier and Type	Method and Description
protected org.springframework.integration.core.MessageProducer	createConsumerEndpoint(ConsumerDestination destination, String group, ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> extendedConsumerProperties) Creates MessageProducer that receives data from the consumer destination.
protected org.springframework.kafka.core.ConsumerFactory<?,?>	createKafkaConsumerFactory(boolean anonymous, String consumerGroup, ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> consumerProperties)
protected AbstractMessageChannelBinder.PolledConsumerResources	createPolledConsumerResources(String name, String group, ConsumerDestination destination, ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> consumerProperties)
protected org.springframework.messaging.MessageHandler	createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties> producerProperties, org.springframework.messaging.MessageChannel errorChannel) Create a MessageHandler with the ability to send data to the target middleware.
protected org.springframework.messaging.MessageHandler	createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties> producerProperties, org.springframework.messaging.MessageChannel channel, org.springframework.messaging.MessageChannel errorChannel) Create a MessageHandler with the ability to send data to the target middleware.
String	getDefaultsPrefix() Extended binding properties can define a default prefix to place all the extended common producer and consumer properties.
protected org.springframework.messaging.MessageHandler	getErrorMessageHandler(ConsumerDestination destination, String group, ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> properties) Binders can return a message handler to be subscribed to the error channel.
protected org.springframework.integration.support.ErrorMessageStrategy	getErrorMessageStrategy() Binders can return an ErrorMessageStrategy for building error messages; binder implementations typically might add extra headers to the error message.
org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties	getExtendedConsumerProperties(String channelName)
org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties	getExtendedProducerProperties(String channelName)
Class<? extends BinderSpecificPropertiesProvider>	getExtendedPropertiesEntryClass() Extended properties class which should be a subclass of BinderSpecificPropertiesProvider against which default extended producer and consumer properties are resolved.
protected org.springframework.messaging.MessageHandler	getPolledConsumerErrorMessageHandler(ConsumerDestination destination, String group, ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> properties) Binders can return a message handler to be subscribed to the error channel.
protected org.springframework.kafka.core.DefaultKafkaProducerFactory<byte[],byte[]>	getProducerFactory(String transactionIdPrefix, ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties> producerProperties)
protected void	postProcessPollableSource(DefaultPollableMessageSource bindingTarget)
Collection<org.apache.kafka.common.PartitionInfo>	processTopic(String group, ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> extendedConsumerProperties, org.springframework.kafka.core.ConsumerFactory<?,?> consumerFactory, int partitionCount, boolean usingPatterns, boolean groupManagement, String topic)
void	setExtendedBindingProperties(org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties extendedBindingProperties)
void	setProducerListener(org.springframework.kafka.support.ProducerListener<byte[],byte[]> producerListener)
protected boolean	useNativeEncoding(ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties> producerProperties) Whether the producer for the destination being created should be configured to use native encoding which may, or may not, be determined from the properties.

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

afterUnbindConsumer, afterUnbindProducer, bindPollableConsumer, doBindConsumer, doBindProducer, errorsBaseName, errorsBaseName, getApplicationEventPublisher, getContainerCustomizer, getDefaultErrorMessageHandler, getErrorBridgeName, getErrorBridgeName, getErrorMessageHandlerName, getErrorRecovererName, getPolledConsumerRecoveryCallback, postProcessOutputChannel, registerErrorInfrastructure, registerErrorInfrastructure, setApplicationEventPublisher

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

afterPropertiesSet, applyPrefix, bindConsumer, bindProducer, buildPartitionRoutingExpression, buildRetryTemplate, constructDLQName, getApplicationContext, getBeanFactory, getEvaluationContext, groupedName, onInit, serializePayloadIfNecessary, setApplicationContext

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

Field Detail

X_EXCEPTION_FQCN

public static final String X_EXCEPTION_FQCN

See Also:

Constant Field Values

X_EXCEPTION_STACKTRACE

public static final String X_EXCEPTION_STACKTRACE

See Also:

Constant Field Values

X_EXCEPTION_MESSAGE

public static final String X_EXCEPTION_MESSAGE

See Also:

Constant Field Values

X_ORIGINAL_TOPIC

public static final String X_ORIGINAL_TOPIC

See Also:

Constant Field Values

X_ORIGINAL_PARTITION

public static final String X_ORIGINAL_PARTITION

See Also:

Constant Field Values

X_ORIGINAL_OFFSET

public static final String X_ORIGINAL_OFFSET

See Also:

Constant Field Values

X_ORIGINAL_TIMESTAMP

public static final String X_ORIGINAL_TIMESTAMP

See Also:

Constant Field Values

X_ORIGINAL_TIMESTAMP_TYPE

public static final String X_ORIGINAL_TIMESTAMP_TYPE

See Also:

Constant Field Values

Constructor Detail

KafkaMessageChannelBinder

public KafkaMessageChannelBinder(org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties configurationProperties,
                                 org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner provisioningProvider)

KafkaMessageChannelBinder

public KafkaMessageChannelBinder(org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties configurationProperties,
                                 org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner provisioningProvider,
                                 ListenerContainerCustomizer<org.springframework.kafka.listener.AbstractMessageListenerContainer<?,?>> containerCustomizer)

Method Detail

setExtendedBindingProperties

public void setExtendedBindingProperties(org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties extendedBindingProperties)

setProducerListener

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

getExtendedConsumerProperties

public org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties getExtendedConsumerProperties(String channelName)

Specified by:

getExtendedConsumerProperties in interface ExtendedBindingProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties,org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>

getExtendedProducerProperties

public org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties getExtendedProducerProperties(String channelName)

Specified by:

getExtendedProducerProperties in interface ExtendedBindingProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties,org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>

getDefaultsPrefix

public String getDefaultsPrefix()

Description copied from interface: ExtendedBindingProperties

Extended binding properties can define a default prefix to place all the extended common producer and consumer properties. For example, if the binder type is foo it is convenient to specify common extended properties for the producer or consumer across multiple bindings in the form of `spring.cloud.stream.foo.default.producer.x=y` or `spring.cloud.stream.foo.default.consumer.x=y`. The binding process will use this defaults prefix to resolve any common extended producer and consumer properties.

Specified by:

getDefaultsPrefix in interface ExtendedBindingProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties,org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>

Returns:

default prefix for extended properties

getExtendedPropertiesEntryClass

public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass()

Description copied from interface: ExtendedBindingProperties

Extended properties class which should be a subclass of BinderSpecificPropertiesProvider against which default extended producer and consumer properties are resolved.

Specified by:

getExtendedPropertiesEntryClass in interface ExtendedBindingProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties,org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>

Returns:

extended properties class that contains extended producer/consumer properties

createProducerMessageHandler

protected org.springframework.messaging.MessageHandler createProducerMessageHandler(ProducerDestination destination,
                                                                                    ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties> producerProperties,
                                                                                    org.springframework.messaging.MessageChannel errorChannel)
                                                                             throws Exception

Description copied from class: AbstractMessageChannelBinder

Create 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<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

Parameters:

destination - the name of the target destination

producerProperties - the producer properties

errorChannel - the error channel (if enabled, otherwise null). If not null, the binder must wire this channel into the producer endpoint so that errors are forwarded to it.

Returns:

the message handler for sending data to the target middleware

Throws:

Exception

createProducerMessageHandler

protected org.springframework.messaging.MessageHandler createProducerMessageHandler(ProducerDestination destination,
                                                                                    ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties> producerProperties,
                                                                                    org.springframework.messaging.MessageChannel channel,
                                                                                    org.springframework.messaging.MessageChannel errorChannel)
                                                                             throws Exception

Description copied from class: AbstractMessageChannelBinder

Create 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

Overrides:

createProducerMessageHandler in class AbstractMessageChannelBinder<ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

Parameters:

destination - the name of the target destination.

producerProperties - the producer properties.

channel - the channel to bind.

errorChannel - the error channel (if enabled, otherwise null). If not null, the binder must wire this channel into the producer endpoint so that errors are forwarded to it.

Returns:

the message handler for sending data to the target middleware

Throws:

Exception

getProducerFactory

protected org.springframework.kafka.core.DefaultKafkaProducerFactory<byte[],byte[]> getProducerFactory(String transactionIdPrefix,
                                                                                                       ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties> producerProperties)

useNativeEncoding

protected boolean useNativeEncoding(ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties> producerProperties)

Description copied from class: AbstractMessageChannelBinder

Whether the producer for the destination being created should be configured to use native encoding which may, or may not, be determined from the properties. For example, a transactional kafka binder uses a common producer for all destinations. The default implementation returns P#isUseNativeEncoding().

Overrides:

useNativeEncoding in class AbstractMessageChannelBinder<ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

Parameters:

producerProperties - the properties.

Returns:

true to use native encoding.

createConsumerEndpoint

protected org.springframework.integration.core.MessageProducer createConsumerEndpoint(ConsumerDestination destination,
                                                                                      String group,
                                                                                      ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> extendedConsumerProperties)

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<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

Parameters:

destination - reference to the consumer destination

group - the consumer group

extendedConsumerProperties - the consumer properties

Returns:

the consumer endpoint.

processTopic

public Collection<org.apache.kafka.common.PartitionInfo> processTopic(String group,
                                                                      ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> extendedConsumerProperties,
                                                                      org.springframework.kafka.core.ConsumerFactory<?,?> consumerFactory,
                                                                      int partitionCount,
                                                                      boolean usingPatterns,
                                                                      boolean groupManagement,
                                                                      String topic)

createPolledConsumerResources

protected AbstractMessageChannelBinder.PolledConsumerResources createPolledConsumerResources(String name,
                                                                                             String group,
                                                                                             ConsumerDestination destination,
                                                                                             ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> consumerProperties)

Overrides:

createPolledConsumerResources in class AbstractMessageChannelBinder<ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

postProcessPollableSource

protected void postProcessPollableSource(DefaultPollableMessageSource bindingTarget)

Overrides:

postProcessPollableSource in class AbstractMessageChannelBinder<ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

getErrorMessageStrategy

protected org.springframework.integration.support.ErrorMessageStrategy getErrorMessageStrategy()

Description copied from class: AbstractMessageChannelBinder

Binders can return an ErrorMessageStrategy for building error messages; binder implementations typically might add extra headers to the error message.

Overrides:

getErrorMessageStrategy in class AbstractMessageChannelBinder<ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

Returns:

the implementation - may be null.

getErrorMessageHandler

protected org.springframework.messaging.MessageHandler getErrorMessageHandler(ConsumerDestination destination,
                                                                              String group,
                                                                              ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> properties)

Description copied from class: AbstractMessageChannelBinder

Binders can return a message handler to be subscribed to the error channel. Examples might be if the user wishes to (re)publish messages to a DLQ.

Overrides:

getErrorMessageHandler in class AbstractMessageChannelBinder<ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

Parameters:

destination - the destination.

group - the group.

properties - the properties.

Returns:

the handler (may be null, which is the default, causing the exception to be rethrown).

getPolledConsumerErrorMessageHandler

protected org.springframework.messaging.MessageHandler getPolledConsumerErrorMessageHandler(ConsumerDestination destination,
                                                                                            String group,
                                                                                            ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> properties)

Description copied from class: AbstractMessageChannelBinder

Binders can return a message handler to be subscribed to the error channel. Examples might be if the user wishes to (re)publish messages to a DLQ.

Overrides:

getPolledConsumerErrorMessageHandler in class AbstractMessageChannelBinder<ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties>,ExtendedProducerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties>,org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner>

Parameters:

destination - the destination.

group - the group.

properties - the properties.

Returns:

the handler (may be null, which is the default, causing the exception to be rethrown).

createKafkaConsumerFactory

protected org.springframework.kafka.core.ConsumerFactory<?,?> createKafkaConsumerFactory(boolean anonymous,
                                                                                         String consumerGroup,
                                                                                         ExtendedConsumerProperties<org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties> consumerProperties)