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>

Direct Known Subclasses:

AggregatingReplyingKafkaTemplate

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/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) Deprecated. in favor of setCorrelationIdStrategy(Function).
void	destroy()
java.util.Collection<org.apache.kafka.common.TopicPartition>	getAssignedReplyTopicPartitions() Return the topics/partitions assigned to the replying listener container.
int	getPhase()
protected long	getReplyTimeout()
protected boolean	handleTimeout(CorrelationKey correlationId, RequestReplyFuture<K,V,R> future) Used to inform subclasses that a request has timed out so they can clean up state and, optionally, complete the future.
boolean	isAutoStartup()
protected boolean	isPending(CorrelationKey correlationId) Return true if this correlation id is still active.
boolean	isRunning()
protected void	logLateArrival(org.apache.kafka.clients.consumer.ConsumerRecord<K,R> record, CorrelationKey correlationId)
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	setCorrelationIdStrategy(java.util.function.Function<org.apache.kafka.clients.producer.ProducerRecord<K,V>,CorrelationKey> correlationStrategy) Set a function to be called to establish a unique correlation key for each request record.
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, getProducerFactory, getTransactionIdPrefix, inTransaction, isTransactional, metrics, partitionsFor, send, send, send, send, send, send, sendDefault, sendDefault, sendDefault, sendDefault, sendOffsetsToTransaction, sendOffsetsToTransaction, setDefaultTopic, setMessageConverter, setProducerListener, setTransactionIdPrefix

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)

getReplyTimeout

protected long getReplyTimeout()

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

setCorrelationIdStrategy

public void setCorrelationIdStrategy(java.util.function.Function<org.apache.kafka.clients.producer.ProducerRecord<K,V>,CorrelationKey> correlationStrategy)

Set a function to be called to establish a unique correlation key for each request record.

Parameters:

correlationStrategy - the function.

Since:

2.3

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.

handleTimeout

protected boolean handleTimeout(CorrelationKey correlationId,
                                RequestReplyFuture<K,V,R> future)

Used to inform subclasses that a request has timed out so they can clean up state and, optionally, complete the future.

Parameters:

correlationId - the correlation id.

future - the future.

Returns:

true to indicate the future has been completed.

Since:

2.3

isPending

protected boolean isPending(CorrelationKey correlationId)

Return true if this correlation id is still active.

Parameters:

correlationId - the correlation id.

Returns:

true if pending.

Since:

2.3

destroy

public void destroy()

Specified by:

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

createCorrelationId

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

Deprecated. in favor of setCorrelationIdStrategy(Function).

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.

logLateArrival

protected void logLateArrival(org.apache.kafka.clients.consumer.ConsumerRecord<K,R> record,
                              CorrelationKey correlationId)