Package org.springframework.integration.kafka.outbound

Class KafkaProducerMessageHandler<K,V>

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.MessageHandlerSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.handler.AbstractMessageProducingHandler

org.springframework.integration.handler.AbstractReplyProducingMessageHandler

org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler<K,V>

Type Parameters:

K - the key type.

V - the value type.

All Implemented Interfaces:

org.reactivestreams.Subscriber<Message<?>>, Aware, BeanClassLoaderAware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, Lifecycle, Ordered, ExpressionCapable, Orderable, MessageProducer, HeaderPropagationAware, IntegrationPattern, NamedComponent, IntegrationManagement, ManageableLifecycle, TrackableComponent, MessageHandler, reactor.core.CoreSubscriber<Message<?>>

public class KafkaProducerMessageHandler<K,V>
extends AbstractReplyProducingMessageHandler
implements ManageableLifecycle

A Message Handler for Apache Kafka; when supplied with a ReplyingKafkaTemplate it is used as the handler in an outbound gateway. When supplied with a simple KafkaTemplate it used as the handler in an outbound channel adapter.

The handler also supports receiving a pre-built ProducerRecord payload. In that case, most configuration properties (setTopicExpression(Expression) etc.) are ignored. If the handler is used as gateway, the ProducerRecord will have its headers enhanced to add the KafkaHeaders.REPLY_TOPIC unless it already contains such a header. The handler will not map any additional headers; providing such a payload assumes the headers have already been mapped.

Since:

5.4

Author:

Soby Chacko, Artem Bilan, Gary Russell, Marius Bogoevici, Biju Kunjummen, Tom van den Berge

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	KafkaProducerMessageHandler.ProducerRecordCreator<K,V>	Creates a ProducerRecord from a Message and/or properties derived from configuration and/or the message.

Nested classes/interfaces inherited from class org.springframework.integration.handler.AbstractReplyProducingMessageHandler

AbstractReplyProducingMessageHandler.RequestHandler

Nested classes/interfaces inherited from interface org.springframework.integration.support.management.IntegrationManagement

IntegrationManagement.ManagementOverrides

Field Summary

Fields inherited from class org.springframework.integration.handler.AbstractMessageProducingHandler

messagingTemplate

Fields inherited from class org.springframework.integration.context.IntegrationObjectSupport

EXPRESSION_PARSER, logger

Fields inherited from interface org.springframework.integration.support.management.IntegrationManagement

METER_PREFIX, RECEIVE_COUNTER_NAME, SEND_TIMER_NAME

Fields inherited from interface org.springframework.core.Ordered

HIGHEST_PRECEDENCE, LOWEST_PRECEDENCE

Constructor Summary

Constructors

Constructor	Description
KafkaProducerMessageHandler(org.springframework.kafka.core.KafkaTemplate<K,V> kafkaTemplate)

Method Summary

Modifier and Type	Method	Description
protected void	doInit()
String	getComponentType()	Subclasses may implement this method to provide component type information.
protected MessageChannel	getFuturesChannel()
org.springframework.kafka.support.KafkaHeaderMapper	getHeaderMapper()
org.springframework.kafka.core.KafkaTemplate<?,?>	getKafkaTemplate()
protected MessageChannel	getSendFailureChannel()
protected MessageChannel	getSendSuccessChannel()
protected Object	handleRequestMessage(Message<?> message)	Subclasses must implement this method to handle the request Message.
boolean	isRunning()
void	processSendResult(Message<?> message, org.apache.kafka.clients.producer.ProducerRecord<K,V> producerRecord, ListenableFuture<org.springframework.kafka.support.SendResult<K,V>> future, MessageChannel metadataChannel)
void	setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy)	Set the error message strategy implementation to use when sending error messages after send failures.
void	setFlushExpression(Expression flushExpression)	Specify a SpEL expression that evaluates to a Boolean to determine whether the producer should be flushed after the send.
void	setFuturesChannel(MessageChannel futuresChannel)	Set the futures channel.
void	setFuturesChannelName(String futuresChannelName)	Set the futures channel name.
void	setHeaderMapper(org.springframework.kafka.support.KafkaHeaderMapper headerMapper)	Set the header mapper to use.
void	setMessageKeyExpression(Expression messageKeyExpression)
void	setPartitionIdExpression(Expression partitionIdExpression)
void	setProducerRecordCreator(KafkaProducerMessageHandler.ProducerRecordCreator<K,V> producerRecordCreator)	Set a KafkaProducerMessageHandler.ProducerRecordCreator to create the ProducerRecord.
void	setReplyMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)	Set a message converter for gateway replies.
void	setReplyPayloadType(Type payloadType)	When using a type-aware message converter (such as StringJsonMessageConverter, set the payload type the converter should create.
void	setSendFailureChannel(MessageChannel sendFailureChannel)	Set the failure channel.
void	setSendFailureChannelName(String sendFailureChannelName)	Set the failure channel name.
void	setSendSuccessChannel(MessageChannel sendSuccessChannel)	Set the success channel.
void	setSendSuccessChannelName(String sendSuccessChannelName)	Set the Success channel name.
void	setSendTimeout(long sendTimeout)	Specify a timeout in milliseconds for how long this KafkaProducerMessageHandler should wait for send operation results.
void	setSendTimeoutExpression(Expression sendTimeoutExpression)	Specify a SpEL expression to evaluate a timeout in milliseconds for how long this KafkaProducerMessageHandler should wait for send operation results.
void	setSync(boolean sync)	A boolean indicating if the KafkaProducerMessageHandler should wait for the send operation results or not.
void	setTimeoutBuffer(int timeoutBuffer)	Set a buffer, in milliseconds, added to the configured delivery.timeout.ms to determine the minimum time to wait for the send future completion when sync is true.
void	setTimestampExpression(Expression timestampExpression)	Specify a SpEL expression to evaluate a timestamp that will be added in the Kafka record.
void	setTopicExpression(Expression topicExpression)
void	setUseTemplateConverter(boolean useTemplateConverter)	Set to true to use the template's message converter to create the ProducerRecord instead of the producerRecordCreator.
void	start()
void	stop()

Methods inherited from class org.springframework.integration.handler.AbstractReplyProducingMessageHandler

doInvokeAdvisedRequestHandler, getBeanClassLoader, getIntegrationPatternType, getRequiresReply, handleMessageInternal, hasAdviceChain, onInit, setAdviceChain, setBeanClassLoader, setRequiresReply

Methods inherited from class org.springframework.integration.handler.AbstractMessageProducingHandler

addNotPropagatedHeaders, createOutputMessage, getNotPropagatedHeaders, getOutputChannel, isAsync, messageBuilderForReply, produceOutput, resolveErrorChannel, sendErrorMessage, sendOutput, sendOutputs, setAsync, setNotPropagatedHeaders, setOutputChannel, setOutputChannelName, shouldCopyRequestHeaders, shouldSplitOutput, updateNotPropagatedHeaders

Methods inherited from class org.springframework.integration.handler.AbstractMessageHandler

handleMessage, onComplete, onError, onNext, onSubscribe

Methods inherited from class org.springframework.integration.handler.MessageHandlerSupport

buildSendTimer, destroy, getManagedName, getManagedType, getMetricsCaptor, getOrder, getOverrides, isLoggingEnabled, registerMetricsCaptor, sendTimer, setLoggingEnabled, setManagedName, setManagedType, setOrder, setShouldTrack, shouldTrack

Methods inherited from class org.springframework.integration.context.IntegrationObjectSupport

afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanDescription, getBeanFactory, getBeanName, getChannelResolver, getComponentName, getConversionService, getExpression, getIntegrationProperties, getIntegrationProperty, getMessageBuilderFactory, getTaskScheduler, isInitialized, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentName, setConversionService, setMessageBuilderFactory, setPrimaryExpression, setTaskScheduler, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface reactor.core.CoreSubscriber

currentContext

Methods inherited from interface org.springframework.integration.support.management.IntegrationManagement

getThisAs

Methods inherited from interface org.springframework.integration.support.context.NamedComponent

getBeanName, getComponentName

Constructor Details

KafkaProducerMessageHandler

public KafkaProducerMessageHandler(org.springframework.kafka.core.KafkaTemplate<K,V> kafkaTemplate)

Method Details

setTopicExpression

public void setTopicExpression(Expression topicExpression)

setMessageKeyExpression

public void setMessageKeyExpression(Expression messageKeyExpression)

setPartitionIdExpression

public void setPartitionIdExpression(Expression partitionIdExpression)

setTimestampExpression

public void setTimestampExpression(Expression timestampExpression)

Specify a SpEL expression to evaluate a timestamp that will be added in the Kafka record. The resulting value should be a Long type representing epoch time in milliseconds.

Parameters:

timestampExpression - the Expression for timestamp to wait for result fo send operation.

setFlushExpression

public void setFlushExpression(Expression flushExpression)

Specify a SpEL expression that evaluates to a Boolean to determine whether the producer should be flushed after the send. Defaults to looking for a Boolean value in a KafkaIntegrationHeaders.FLUSH header; false if absent.

Parameters:

flushExpression - the Expression.

setHeaderMapper

public void setHeaderMapper(org.springframework.kafka.support.KafkaHeaderMapper headerMapper)

Set the header mapper to use.

Parameters:

headerMapper - the mapper; can be null to disable header mapping.

getHeaderMapper

public org.springframework.kafka.support.KafkaHeaderMapper getHeaderMapper()

getKafkaTemplate

public org.springframework.kafka.core.KafkaTemplate<?,?> getKafkaTemplate()

setSync

public void setSync(boolean sync)

A boolean indicating if the KafkaProducerMessageHandler should wait for the send operation results or not. Defaults to false. In sync mode a downstream send operation exception will be re-thrown.

Parameters:

sync - the send mode; async by default.

setSendTimeout

public final void setSendTimeout(long sendTimeout)

Specify a timeout in milliseconds for how long this KafkaProducerMessageHandler should wait for send operation results. Defaults to the kafka delivery.timeout.ms property + 5 seconds. The timeout is applied Also applies when sending to the success or failure channels.

Overrides:

setSendTimeout in class AbstractMessageProducingHandler

Parameters:

sendTimeout - the timeout to wait for result for a send operation.

setSendTimeoutExpression

public void setSendTimeoutExpression(Expression sendTimeoutExpression)

Specify a SpEL expression to evaluate a timeout in milliseconds for how long this KafkaProducerMessageHandler should wait for send operation results. Defaults to the kafka delivery.timeout.ms property + 5 seconds. The timeout is applied only in sync mode. If this expression yields a result that is less than that value, the higher value is used.

Parameters:

sendTimeoutExpression - the Expression for timeout to wait for result for a send operation.

See Also:

setTimeoutBuffer(int)

setSendFailureChannel

public void setSendFailureChannel(MessageChannel sendFailureChannel)

Set the failure channel. After a send failure, an ErrorMessage will be sent to this channel with a payload of a KafkaSendFailureException with the failed message and cause.

Parameters:

sendFailureChannel - the failure channel.

setSendFailureChannelName

public void setSendFailureChannelName(String sendFailureChannelName)

Set the failure channel name. After a send failure, an ErrorMessage will be sent to this channel name with a payload of a KafkaSendFailureException with the failed message and cause.

Parameters:

sendFailureChannelName - the failure channel name.

setSendSuccessChannel

public void setSendSuccessChannel(MessageChannel sendSuccessChannel)

Set the success channel.

Parameters:

sendSuccessChannel - the Success channel.

setSendSuccessChannelName

public void setSendSuccessChannelName(String sendSuccessChannelName)

Set the Success channel name.

Parameters:

sendSuccessChannelName - the Success channel name.

setFuturesChannel

public void setFuturesChannel(MessageChannel futuresChannel)

Set the futures channel.

Parameters:

futuresChannel - the futures channel.

setFuturesChannelName

public void setFuturesChannelName(String futuresChannelName)

Set the futures channel name.

Parameters:

futuresChannelName - the futures channel name.

setErrorMessageStrategy

public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy)

Set the error message strategy implementation to use when sending error messages after send failures. Cannot be null.

Parameters:

errorMessageStrategy - the implementation.

setReplyMessageConverter

public void setReplyMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)

Set a message converter for gateway replies.

Parameters:

messageConverter - the converter.

See Also:

setReplyPayloadType(Type)

setReplyPayloadType

public void setReplyPayloadType(Type payloadType)

When using a type-aware message converter (such as StringJsonMessageConverter, set the payload type the converter should create. Defaults to Object.

Parameters:

payloadType - the type.

See Also:

setReplyMessageConverter(RecordMessageConverter)

setProducerRecordCreator

public void setProducerRecordCreator(KafkaProducerMessageHandler.ProducerRecordCreator<K,V> producerRecordCreator)

Set a KafkaProducerMessageHandler.ProducerRecordCreator to create the ProducerRecord. Ignored if useTemplateConverter is true.

Parameters:

producerRecordCreator - the creator.

See Also:

setUseTemplateConverter(boolean)

setTimeoutBuffer

public void setTimeoutBuffer(int timeoutBuffer)

Set a buffer, in milliseconds, added to the configured delivery.timeout.ms to determine the minimum time to wait for the send future completion when sync is true.

Parameters:

timeoutBuffer - the buffer.

See Also:

setSendTimeoutExpression(Expression)

setUseTemplateConverter

public void setUseTemplateConverter(boolean useTemplateConverter)

Set to true to use the template's message converter to create the ProducerRecord instead of the producerRecordCreator.

Parameters:

useTemplateConverter - true to use the converter.

Since:

5.5.5

See Also:

setProducerRecordCreator(ProducerRecordCreator)

getComponentType

public String getComponentType()

Description copied from class: IntegrationObjectSupport

Subclasses may implement this method to provide component type information.

Specified by:

getComponentType in interface NamedComponent

Overrides:

getComponentType in class MessageHandlerSupport

getSendFailureChannel

protected MessageChannel getSendFailureChannel()

getSendSuccessChannel

protected MessageChannel getSendSuccessChannel()

getFuturesChannel

protected MessageChannel getFuturesChannel()

doInit

protected void doInit()

Overrides:

doInit in class AbstractReplyProducingMessageHandler

start

public void start()

Specified by:

start in interface Lifecycle

Specified by:

start in interface ManageableLifecycle

stop

public void stop()

Specified by:

stop in interface Lifecycle

Specified by:

stop in interface ManageableLifecycle

isRunning

public boolean isRunning()

Specified by:

isRunning in interface Lifecycle

Specified by:

isRunning in interface ManageableLifecycle

handleRequestMessage

protected Object handleRequestMessage(Message<?> message)

Description copied from class: AbstractReplyProducingMessageHandler

Subclasses must implement this method to handle the request Message. The return value may be a Message, a MessageBuilder, or any plain Object. The base class will handle the final creation of a reply Message from any of those starting points. If the return value is null, the Message flow will end here.

Specified by:

handleRequestMessage in class AbstractReplyProducingMessageHandler

Parameters:

message - The request message.

Returns:

The result of handling the message, or null.

processSendResult

public void processSendResult(Message<?> message, org.apache.kafka.clients.producer.ProducerRecord<K,V> producerRecord, ListenableFuture<org.springframework.kafka.support.SendResult<K,V>> future, MessageChannel metadataChannel) throws InterruptedException, ExecutionException

Throws:

InterruptedException

ExecutionException