org.springframework.kafka.requestreply

Class ReplyingKafkaTemplate<K,V,R>

java.lang.Object

org.springframework.kafka.core.KafkaTemplate<K,V>

org.springframework.kafka.requestreply.ReplyingKafkaTemplate<K,V,R>

Type Parameters:

K - the key type.

V - the outbound data type.

R - the reply data type.

All Implemented Interfaces:

org.springframework.beans.factory.DisposableBean, org.springframework.beans.factory.InitializingBean, org.springframework.context.Lifecycle, org.springframework.context.Phased, org.springframework.context.SmartLifecycle, KafkaOperations<K,V>, BatchMessageListener<K,R>, GenericMessageListener<java.util.List<org.apache.kafka.clients.consumer.ConsumerRecord<K,R>>>, ReplyingKafkaOperations<K,V,R>

public class ReplyingKafkaTemplate<K,V,R>
extends KafkaTemplate<K,V>
implements BatchMessageListener<K,R>, org.springframework.beans.factory.InitializingBean, org.springframework.context.SmartLifecycle, org.springframework.beans.factory.DisposableBean, ReplyingKafkaOperations<K,V,R>

A KafkaTemplate that implements request/reply semantics.

Since:

2.1.3

Author:

Gary Russell

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	ReplyingKafkaTemplate.TemplateRequestReplyFuture<K,V,R> A listenable future for requests/replies.

Nested classes/interfaces inherited from interface org.springframework.kafka.core.KafkaOperations

KafkaOperations.OperationsCallback<K,V,T>, KafkaOperations.ProducerCallback<K,V,T>

Field Summary

Fields inherited from class org.springframework.kafka.core.KafkaTemplate

logger

Fields inherited from interface org.springframework.context.SmartLifecycle

DEFAULT_PHASE

Constructor Summary

Constructors

Constructor and Description
ReplyingKafkaTemplate(ProducerFactory<K,V> producerFactory, GenericMessageListenerContainer<K,R> replyContainer)
ReplyingKafkaTemplate(ProducerFactory<K,V> producerFactory, GenericMessageListenerContainer<K,R> replyContainer, boolean autoFlush)

Method Summary

Modifier and Type	Method and Description
void	afterPropertiesSet()
protected CorrelationKey	createCorrelationId(org.apache.kafka.clients.producer.ProducerRecord<K,V> record) Subclasses can override this to generate custom correlation ids.
void	destroy()
java.util.Collection<org.apache.kafka.common.TopicPartition>	getAssignedReplyTopicPartitions() Return the topics/partitions assigned to the replying listener container.
int	getPhase()
boolean	isAutoStartup()
boolean	isRunning()
void	onMessage(java.util.List<org.apache.kafka.clients.consumer.ConsumerRecord<K,R>> data) Invoked with data from kafka.
RequestReplyFuture<K,V,R>	sendAndReceive(org.apache.kafka.clients.producer.ProducerRecord<K,V> record) Send a request and receive a reply.
void	setAutoStartup(boolean autoStartup)
void	setPhase(int phase)
void	setReplyTimeout(long replyTimeout)
void	setSharedReplyTopic(boolean sharedReplyTopic) Set to true when multiple templates are using the same topic for replies.
void	setTaskScheduler(org.springframework.scheduling.TaskScheduler scheduler)
void	start()
void	stop()
void	stop(java.lang.Runnable callback)

Methods inherited from class org.springframework.kafka.core.KafkaTemplate

closeProducer, doSend, execute, executeInTransaction, flush, getDefaultTopic, getMessageConverter, inTransaction, isTransactional, metrics, partitionsFor, send, send, send, send, send, send, sendDefault, sendDefault, sendDefault, sendDefault, sendOffsetsToTransaction, sendOffsetsToTransaction, setDefaultTopic, setMessageConverter, setProducerListener

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.kafka.listener.BatchMessageListener

onMessage, wantsPollResult

Methods inherited from interface org.springframework.kafka.listener.GenericMessageListener

onMessage, onMessage, onMessage

Constructor Detail

ReplyingKafkaTemplate

public ReplyingKafkaTemplate(ProducerFactory<K,V> producerFactory,
                             GenericMessageListenerContainer<K,R> replyContainer)

ReplyingKafkaTemplate

public ReplyingKafkaTemplate(ProducerFactory<K,V> producerFactory,
                             GenericMessageListenerContainer<K,R> replyContainer,
                             boolean autoFlush)

Method Detail

setTaskScheduler

public void setTaskScheduler(org.springframework.scheduling.TaskScheduler scheduler)

setReplyTimeout

public void setReplyTimeout(long replyTimeout)

isRunning

public boolean isRunning()

Specified by:

isRunning in interface org.springframework.context.Lifecycle

getPhase

public int getPhase()

Specified by:

getPhase in interface org.springframework.context.Phased

Specified by:

getPhase in interface org.springframework.context.SmartLifecycle

setPhase

public void setPhase(int phase)

isAutoStartup

public boolean isAutoStartup()

Specified by:

isAutoStartup in interface org.springframework.context.SmartLifecycle

setAutoStartup

public void setAutoStartup(boolean autoStartup)

getAssignedReplyTopicPartitions

public java.util.Collection<org.apache.kafka.common.TopicPartition> getAssignedReplyTopicPartitions()

Return the topics/partitions assigned to the replying listener container.

Returns:

the topics/partitions.

setSharedReplyTopic

public void setSharedReplyTopic(boolean sharedReplyTopic)

Set to true when multiple templates are using the same topic for replies. This simply changes logs for unexpected replies to debug instead of error.

Parameters:

sharedReplyTopic - true if using a shared topic.

Since:

2.2

afterPropertiesSet

public void afterPropertiesSet()

Specified by:

afterPropertiesSet in interface org.springframework.beans.factory.InitializingBean

start

public void start()

Specified by:

start in interface org.springframework.context.Lifecycle

stop

public void stop()

Specified by:

stop in interface org.springframework.context.Lifecycle

stop

public void stop(java.lang.Runnable callback)

Specified by:

stop in interface org.springframework.context.SmartLifecycle

sendAndReceive

public RequestReplyFuture<K,V,R> sendAndReceive(org.apache.kafka.clients.producer.ProducerRecord<K,V> record)

Description copied from interface: ReplyingKafkaOperations

Send a request and receive a reply.

Specified by:

sendAndReceive in interface ReplyingKafkaOperations<K,V,R>

Parameters:

record - the record to send.

Returns:

a RequestReplyFuture.

destroy

public void destroy()

Specified by:

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

createCorrelationId

protected CorrelationKey createCorrelationId(org.apache.kafka.clients.producer.ProducerRecord<K,V> record)

Subclasses can override this to generate custom correlation ids. The default implementation is a 16 byte representation of a UUID.

Parameters:

record - the record.

Returns:

the key.

onMessage

public void onMessage(java.util.List<org.apache.kafka.clients.consumer.ConsumerRecord<K,R>> data)

Description copied from interface: GenericMessageListener

Invoked with data from kafka.

Specified by:

onMessage in interface GenericMessageListener<java.util.List<org.apache.kafka.clients.consumer.ConsumerRecord<K,R>>>

Parameters:

data - the data to be processed.