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<org.springframework.messaging.Message<?>>,org.springframework.beans.factory.Aware,org.springframework.beans.factory.BeanClassLoaderAware,org.springframework.beans.factory.BeanFactoryAware,org.springframework.beans.factory.BeanNameAware,org.springframework.beans.factory.DisposableBean,org.springframework.beans.factory.InitializingBean,org.springframework.context.ApplicationContextAware,org.springframework.context.Lifecycle,org.springframework.core.Ordered,org.springframework.integration.context.ExpressionCapable,org.springframework.integration.context.Orderable,org.springframework.integration.core.MessageProducer,org.springframework.integration.handler.HeaderPropagationAware,org.springframework.integration.IntegrationPattern,org.springframework.integration.support.context.NamedComponent,org.springframework.integration.support.management.IntegrationManagement,org.springframework.integration.support.management.ManageableLifecycle,org.springframework.integration.support.management.TrackableComponent,org.springframework.messaging.MessageHandler,reactor.core.CoreSubscriber<org.springframework.messaging.Message<?>>
public class KafkaProducerMessageHandler<K,V>
extends org.springframework.integration.handler.AbstractReplyProducingMessageHandler
implements org.springframework.integration.support.management.ManageableLifecycle
Kafka Message Handler; 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.
Starting with version 3.2.1 the handler 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
-
Nested Class Summary
Nested Classes Modifier and Type Class Description static interfaceKafkaProducerMessageHandler.ProducerRecordCreator<K,V>Creates aProducerRecordfrom aMessageand/or properties derived from configuration and/or the message. -
Field Summary
Fields inherited from class org.springframework.integration.handler.AbstractMessageProducingHandler
messagingTemplateFields inherited from class org.springframework.integration.context.IntegrationObjectSupport
EXPRESSION_PARSER, logger -
Constructor Summary
Constructors Constructor Description KafkaProducerMessageHandler(org.springframework.kafka.core.KafkaTemplate<K,V> kafkaTemplate) -
Method Summary
Modifier and Type Method Description protected voiddoInit()java.lang.StringgetComponentType()protected org.springframework.messaging.MessageChannelgetFuturesChannel()org.springframework.kafka.support.KafkaHeaderMappergetHeaderMapper()org.springframework.kafka.core.KafkaTemplate<?,?>getKafkaTemplate()protected org.springframework.messaging.MessageChannelgetSendFailureChannel()protected org.springframework.messaging.MessageChannelgetSendSuccessChannel()protected java.lang.ObjecthandleRequestMessage(org.springframework.messaging.Message<?> message)booleanisRunning()voidprocessSendResult(org.springframework.messaging.Message<?> message, org.apache.kafka.clients.producer.ProducerRecord<K,V> producerRecord, org.springframework.util.concurrent.ListenableFuture<org.springframework.kafka.support.SendResult<K,V>> future, org.springframework.messaging.MessageChannel metadataChannel)voidsetErrorMessageStrategy(org.springframework.integration.support.ErrorMessageStrategy errorMessageStrategy)Set the error message strategy implementation to use when sending error messages after send failures.voidsetFlushExpression(org.springframework.expression.Expression flushExpression)Specify a SpEL expression that evaluates to aBooleanto determine whether the producer should be flushed after the send.voidsetFuturesChannel(org.springframework.messaging.MessageChannel futuresChannel)Set the futures channel.voidsetFuturesChannelName(java.lang.String futuresChannelName)Set the futures channel name.voidsetHeaderMapper(org.springframework.kafka.support.KafkaHeaderMapper headerMapper)Set the header mapper to use.voidsetMessageKeyExpression(org.springframework.expression.Expression messageKeyExpression)voidsetPartitionIdExpression(org.springframework.expression.Expression partitionIdExpression)voidsetProducerRecordCreator(KafkaProducerMessageHandler.ProducerRecordCreator<K,V> producerRecordCreator)Set aKafkaProducerMessageHandler.ProducerRecordCreatorto create theProducerRecord.voidsetReplyMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)Set a message converter for gateway replies.voidsetReplyPayloadType(java.lang.reflect.Type payloadType)When using a type-aware message converter (such asStringJsonMessageConverter, set the payload type the converter should create.voidsetSendFailureChannel(org.springframework.messaging.MessageChannel sendFailureChannel)Set the failure channel.voidsetSendFailureChannelName(java.lang.String sendFailureChannelName)Set the failure channel name.voidsetSendSuccessChannel(org.springframework.messaging.MessageChannel sendSuccessChannel)Set the success channel.voidsetSendSuccessChannelName(java.lang.String sendSuccessChannelName)Set the Success channel name.voidsetSendTimeout(long sendTimeout)Specify a timeout in milliseconds for how long thisKafkaProducerMessageHandlershould wait wait for send operation results.voidsetSendTimeoutExpression(org.springframework.expression.Expression sendTimeoutExpression)Specify a SpEL expression to evaluate a timeout in milliseconds for how long thisKafkaProducerMessageHandlershould wait wait for send operation results.voidsetSync(boolean sync)Abooleanindicating if theKafkaProducerMessageHandlershould wait for the send operation results or not.voidsetTimeoutBuffer(int timeoutBuffer)Set a buffer, in milliseconds, added to the configureddelivery.timeout.msto determine the minimum time to wait for the send future completion whensyncis true.voidsetTimestampExpression(org.springframework.expression.Expression timestampExpression)Specify a SpEL expression to evaluate a timestamp that will be added in the Kafka record.voidsetTopicExpression(org.springframework.expression.Expression topicExpression)voidstart()voidstop()Methods inherited from class org.springframework.integration.handler.AbstractReplyProducingMessageHandler
doInvokeAdvisedRequestHandler, getBeanClassLoader, getIntegrationPatternType, getRequiresReply, handleMessageInternal, hasAdviceChain, onInit, setAdviceChain, setBeanClassLoader, setRequiresReplyMethods 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, updateNotPropagatedHeadersMethods inherited from class org.springframework.integration.handler.AbstractMessageHandler
handleMessage, onComplete, onError, onNext, onSubscribeMethods inherited from class org.springframework.integration.handler.MessageHandlerSupport
buildSendTimer, destroy, getManagedName, getManagedType, getMetricsCaptor, getOrder, getOverrides, isLoggingEnabled, registerMetricsCaptor, sendTimer, setLoggingEnabled, setManagedName, setManagedType, setOrder, setShouldTrack, shouldTrackMethods 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, toStringMethods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
-
Constructor Details
-
Method Details
-
setTopicExpression
public void setTopicExpression(org.springframework.expression.Expression topicExpression) -
setMessageKeyExpression
public void setMessageKeyExpression(org.springframework.expression.Expression messageKeyExpression) -
setPartitionIdExpression
public void setPartitionIdExpression(org.springframework.expression.Expression partitionIdExpression) -
setTimestampExpression
public void setTimestampExpression(org.springframework.expression.Expression timestampExpression)Specify a SpEL expression to evaluate a timestamp that will be added in the Kafka record. The resulting value should be aLongtype representing epoch time in milliseconds.- Parameters:
timestampExpression- theExpressionfor timestamp to wait for result fo send operation.- Since:
- 2.3
-
setFlushExpression
public void setFlushExpression(org.springframework.expression.Expression flushExpression)Specify a SpEL expression that evaluates to aBooleanto determine whether the producer should be flushed after the send. Defaults to looking for aBooleanvalue in aKafkaIntegrationHeaders.FLUSHheader; false if absent.- Parameters:
flushExpression- theExpression.- Since:
- 3.3
-
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.- Since:
- 2.3
-
getHeaderMapper
public org.springframework.kafka.support.KafkaHeaderMapper getHeaderMapper() -
getKafkaTemplate
public org.springframework.kafka.core.KafkaTemplate<?,?> getKafkaTemplate() -
setSync
public void setSync(boolean sync)Abooleanindicating if theKafkaProducerMessageHandlershould wait for the send operation results or not. Defaults tofalse. Insyncmode a downstream send operation exception will be re-thrown.- Parameters:
sync- the send mode; async by default.- Since:
- 2.0.1
-
setSendTimeout
public final void setSendTimeout(long sendTimeout)Specify a timeout in milliseconds for how long thisKafkaProducerMessageHandlershould wait wait for send operation results. Defaults to the kafkadelivery.timeout.msproperty + 5 seconds. The timeout is applied Also applies when sending to the success or failure channels.- Overrides:
setSendTimeoutin classorg.springframework.integration.handler.AbstractMessageProducingHandler- Parameters:
sendTimeout- the timeout to wait for result for a send operation.- Since:
- 2.0.1
-
setSendTimeoutExpression
public void setSendTimeoutExpression(org.springframework.expression.Expression sendTimeoutExpression)Specify a SpEL expression to evaluate a timeout in milliseconds for how long thisKafkaProducerMessageHandlershould wait wait for send operation results. Defaults to the kafkadelivery.timeout.msproperty + 5 seconds. The timeout is applied only insyncmode. If this expression yields a result that is less than that value, the higher value is used.- Parameters:
sendTimeoutExpression- theExpressionfor timeout to wait for result for a send operation.- Since:
- 2.1.1
- See Also:
setTimeoutBuffer(int)
-
setSendFailureChannel
public void setSendFailureChannel(org.springframework.messaging.MessageChannel sendFailureChannel)Set the failure channel. After a send failure, anErrorMessagewill be sent to this channel with a payload of aKafkaSendFailureExceptionwith the failed message and cause.- Parameters:
sendFailureChannel- the failure channel.- Since:
- 2.1.2
-
setSendFailureChannelName
public void setSendFailureChannelName(java.lang.String sendFailureChannelName)Set the failure channel name. After a send failure, anErrorMessagewill be sent to this channel name with a payload of aKafkaSendFailureExceptionwith the failed message and cause.- Parameters:
sendFailureChannelName- the failure channel name.- Since:
- 2.1.2
-
setSendSuccessChannel
public void setSendSuccessChannel(org.springframework.messaging.MessageChannel sendSuccessChannel)Set the success channel.- Parameters:
sendSuccessChannel- the Success channel.- Since:
- 3.0.2
-
setSendSuccessChannelName
public void setSendSuccessChannelName(java.lang.String sendSuccessChannelName)Set the Success channel name.- Parameters:
sendSuccessChannelName- the Success channel name.- Since:
- 3.0.2
-
setFuturesChannel
public void setFuturesChannel(org.springframework.messaging.MessageChannel futuresChannel)Set the futures channel.- Parameters:
futuresChannel- the futures channel.- Since:
- 5.4
-
setFuturesChannelName
public void setFuturesChannelName(java.lang.String futuresChannelName)Set the futures channel name.- Parameters:
futuresChannelName- the futures channel name.- Since:
- 5.4
-
setErrorMessageStrategy
public void setErrorMessageStrategy(org.springframework.integration.support.ErrorMessageStrategy errorMessageStrategy)Set the error message strategy implementation to use when sending error messages after send failures. Cannot be null.- Parameters:
errorMessageStrategy- the implementation.- Since:
- 2.1.2
-
setReplyMessageConverter
public void setReplyMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)Set a message converter for gateway replies.- Parameters:
messageConverter- the converter.- Since:
- 3.0.2
- See Also:
setReplyPayloadType(Type)
-
setReplyPayloadType
public void setReplyPayloadType(java.lang.reflect.Type payloadType)When using a type-aware message converter (such asStringJsonMessageConverter, set the payload type the converter should create. Defaults toObject.- Parameters:
payloadType- the type.- Since:
- 3.0.2
- See Also:
setReplyMessageConverter(RecordMessageConverter)
-
setProducerRecordCreator
public void setProducerRecordCreator(KafkaProducerMessageHandler.ProducerRecordCreator<K,V> producerRecordCreator)Set aKafkaProducerMessageHandler.ProducerRecordCreatorto create theProducerRecord.- Parameters:
producerRecordCreator- the creator.- Since:
- 3.2.1
-
setTimeoutBuffer
public void setTimeoutBuffer(int timeoutBuffer)Set a buffer, in milliseconds, added to the configureddelivery.timeout.msto determine the minimum time to wait for the send future completion whensyncis true.- Parameters:
timeoutBuffer- the buffer.- Since:
- 5.4
- See Also:
setSendTimeoutExpression(Expression)
-
getComponentType
public java.lang.String getComponentType()- Specified by:
getComponentTypein interfaceorg.springframework.integration.support.context.NamedComponent- Overrides:
getComponentTypein classorg.springframework.integration.handler.MessageHandlerSupport
-
getSendFailureChannel
protected org.springframework.messaging.MessageChannel getSendFailureChannel() -
getSendSuccessChannel
protected org.springframework.messaging.MessageChannel getSendSuccessChannel() -
getFuturesChannel
protected org.springframework.messaging.MessageChannel getFuturesChannel() -
doInit
protected void doInit()- Overrides:
doInitin classorg.springframework.integration.handler.AbstractReplyProducingMessageHandler
-
start
public void start()- Specified by:
startin interfaceorg.springframework.context.Lifecycle- Specified by:
startin interfaceorg.springframework.integration.support.management.ManageableLifecycle
-
stop
public void stop()- Specified by:
stopin interfaceorg.springframework.context.Lifecycle- Specified by:
stopin interfaceorg.springframework.integration.support.management.ManageableLifecycle
-
isRunning
public boolean isRunning()- Specified by:
isRunningin interfaceorg.springframework.context.Lifecycle- Specified by:
isRunningin interfaceorg.springframework.integration.support.management.ManageableLifecycle
-
handleRequestMessage
protected java.lang.Object handleRequestMessage(org.springframework.messaging.Message<?> message)- Specified by:
handleRequestMessagein classorg.springframework.integration.handler.AbstractReplyProducingMessageHandler
-
processSendResult
public void processSendResult(org.springframework.messaging.Message<?> message, org.apache.kafka.clients.producer.ProducerRecord<K,V> producerRecord, org.springframework.util.concurrent.ListenableFuture<org.springframework.kafka.support.SendResult<K,V>> future, org.springframework.messaging.MessageChannel metadataChannel) throws java.lang.InterruptedException, java.util.concurrent.ExecutionException- Throws:
java.lang.InterruptedExceptionjava.util.concurrent.ExecutionException
-