org.springframework.integration.kafka.dsl

Class KafkaProducerMessageHandlerSpec<K,V,S extends KafkaProducerMessageHandlerSpec<K,V,S>>

java.lang.Object

org.springframework.beans.factory.config.AbstractFactoryBean<T>

org.springframework.integration.dsl.IntegrationComponentSpec<S,H>

org.springframework.integration.dsl.MessageHandlerSpec<S,KafkaProducerMessageHandler<K,V>>

org.springframework.integration.kafka.dsl.KafkaProducerMessageHandlerSpec<K,V,S>

Type Parameters:

K - the key type.

V - the value type.

S - the KafkaProducerMessageHandlerSpec extension type.

All Implemented Interfaces:

Aware, BeanClassLoaderAware, BeanFactoryAware, DisposableBean, FactoryBean<KafkaProducerMessageHandler<K,V>>, InitializingBean, Lifecycle, Phased, SmartLifecycle

Direct Known Subclasses:

KafkaOutboundGatewaySpec, KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec

public class KafkaProducerMessageHandlerSpec<K,V,S extends KafkaProducerMessageHandlerSpec<K,V,S>>
extends MessageHandlerSpec<S,KafkaProducerMessageHandler<K,V>>

A MessageHandlerSpec implementation for the KafkaProducerMessageHandler.

Since:

5.4

Author:

Artem Bilan, Biju Kunjummen, Gary Russell

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<K,V> A KafkaTemplate-based KafkaProducerMessageHandlerSpec extension.

Field Summary

Fields inherited from class org.springframework.integration.dsl.IntegrationComponentSpec

PARSER, target

Fields inherited from class org.springframework.beans.factory.config.AbstractFactoryBean

logger

Fields inherited from interface org.springframework.context.SmartLifecycle

DEFAULT_PHASE

Fields inherited from interface org.springframework.beans.factory.FactoryBean

OBJECT_TYPE_ATTRIBUTE

Method Summary

Modifier and Type	Method and Description
<P> S	flush(java.util.function.Function<Message<P>,Boolean> flushFunction) Configure a Function that will be invoked at runtime to determine whether or not to flush the producer after a send.
S	flushExpression(Expression flushExpression) Configure an Expression to determine whether or not to flush the producer after a send.
S	flushExpression(String flushExpression) Configure a SpEL expression to determine whether or not to flush the producer after a send.
S	futuresChannel(MessageChannel futuresChannel) Set the channel to which send futures are sent.
S	futuresChannel(String futuresChannel) Set the channel to which send futures are sent.
S	headerMapper(org.springframework.kafka.support.KafkaHeaderMapper mapper) Specify a header mapper to map spring messaging headers to Kafka headers.
<P> S	messageKey(java.util.function.Function<Message<P>,?> messageKeyFunction) Configure a Function that will be invoked at runtime to determine the message key under which a message will be stored in the topic.
S	messageKey(String messageKey) Configure the message key to store message in Kafka topic.
S	messageKeyExpression(Expression messageKeyExpression) Configure an Expression to determine the Kafka message key to store at runtime against request Message as a root object of evaluation context.
S	messageKeyExpression(String messageKeyExpression) Configure a SpEL expression to determine the Kafka message key to store at runtime against request Message as a root object of evaluation context.
<P> S	partitionId(java.util.function.Function<Message<P>,Integer> partitionIdFunction) Configure a Function that will be invoked at runtime to determine the partition id under which a message will be stored in the topic.
S	partitionId(Integer partitionId) Configure a partitionId of Kafka topic.
S	partitionIdExpression(Expression partitionIdExpression) Configure an Expression to determine the topic partitionId at runtime against request Message as a root object of evaluation context.
S	partitionIdExpression(String partitionIdExpression) Configure a SpEL expression to determine the topic partitionId at runtime against request Message as a root object of evaluation context.
S	sendFailureChannel(MessageChannel sendFailureChannel) Set the channel to which failed send results are sent.
S	sendFailureChannel(String sendFailureChannel) Set the channel to which failed send results are sent.
S	sendSuccessChannel(MessageChannel sendSuccessChannel) Set the channel to which successful send results are sent.
S	sendSuccessChannel(String sendSuccessChannel) Set the channel to which successful send results are sent.
S	sendTimeout(long sendTimeout) Specify a timeout in milliseconds how long KafkaProducerMessageHandler should wait wait for send operation results.
S	sync(boolean sync) A boolean indicating if the KafkaProducerMessageHandler should wait for the send operation results or not.
<P> S	timestamp(java.util.function.Function<Message<P>,Long> timestampFunction) Configure a Function that will be invoked at runtime to determine the Kafka record timestamp will be stored in the topic.
S	timestampExpression(Expression timestampExpression) Configure an Expression to determine the timestamp at runtime against a request Message as a root object of evaluation context.
S	timestampExpression(String timestampExpression) Configure a SpEL expression to determine the timestamp at runtime against a request Message as a root object of evaluation context.
<P> S	topic(java.util.function.Function<Message<P>,String> topicFunction) Configure a Function that will be invoked at runtime to determine the topic to which a message will be sent.
S	topic(String topic) Configure the Kafka topic to send messages.
S	topicExpression(Expression topicExpression) Configure an Expression to determine the Kafka topic at runtime against request Message as a root object of evaluation context.
S	topicExpression(String topicExpression) Configure a SpEL expression to determine the Kafka topic at runtime against request Message as a root object of evaluation context.

Methods inherited from class org.springframework.integration.dsl.IntegrationComponentSpec

_this, createInstance, destroyInstance, doGet, get, getId, getObjectType, getPhase, id, isAutoStartup, isRunning, start, stop, stop

Methods inherited from class org.springframework.beans.factory.config.AbstractFactoryBean

afterPropertiesSet, destroy, getBeanFactory, getBeanTypeConverter, getEarlySingletonInterfaces, getObject, isSingleton, setBeanClassLoader, setBeanFactory, setSingleton

Methods inherited from class java.lang.Object

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

Method Detail

topic

public S topic(String topic)

Configure the Kafka topic to send messages.

Parameters:

topic - the Kafka topic name.

Returns:

the spec.

topicExpression

public S topicExpression(String topicExpression)

Configure a SpEL expression to determine the Kafka topic at runtime against request Message as a root object of evaluation context.

Parameters:

topicExpression - the topic SpEL expression.

Returns:

the spec.

topicExpression

public S topicExpression(Expression topicExpression)

Configure an Expression to determine the Kafka topic at runtime against request Message as a root object of evaluation context.

Parameters:

topicExpression - the topic expression.

Returns:

the spec.

topic

public <P> S topic(java.util.function.Function<Message<P>,String> topicFunction)

Configure a Function that will be invoked at runtime to determine the topic to which a message will be sent. Typically used with a Java 8 Lambda expression:

 .<Foo>topic(m -> m.getPayload().getTopic())
 
 

Type Parameters:

P - the expected payload type.

Parameters:

topicFunction - the topic function.

Returns:

the current KafkaProducerMessageHandlerSpec.

See Also:

FunctionExpression

messageKeyExpression

public S messageKeyExpression(String messageKeyExpression)

Configure a SpEL expression to determine the Kafka message key to store at runtime against request Message as a root object of evaluation context.

Parameters:

messageKeyExpression - the message key SpEL expression.

Returns:

the spec.

messageKey

public S messageKey(String messageKey)

Configure the message key to store message in Kafka topic.

Parameters:

messageKey - the message key to use.

Returns:

the spec.

messageKeyExpression

public S messageKeyExpression(Expression messageKeyExpression)

Configure an Expression to determine the Kafka message key to store at runtime against request Message as a root object of evaluation context.

Parameters:

messageKeyExpression - the message key expression.

Returns:

the spec.

messageKey

public <P> S messageKey(java.util.function.Function<Message<P>,?> messageKeyFunction)

Configure a Function that will be invoked at runtime to determine the message key under which a message will be stored in the topic. Typically used with a Java 8 Lambda expression:

 .<Foo>messageKey(m -> m.getPayload().getKey())
 
 

Type Parameters:

P - the expected payload type.

Parameters:

messageKeyFunction - the message key function.

Returns:

the current KafkaProducerMessageHandlerSpec.

See Also:

FunctionExpression

partitionId

public S partitionId(Integer partitionId)

Configure a partitionId of Kafka topic.

Parameters:

partitionId - the partitionId to use.

Returns:

the spec.

partitionIdExpression

public S partitionIdExpression(String partitionIdExpression)

Configure a SpEL expression to determine the topic partitionId at runtime against request Message as a root object of evaluation context.

Parameters:

partitionIdExpression - the partitionId expression to use.

Returns:

the spec.

partitionId

public <P> S partitionId(java.util.function.Function<Message<P>,Integer> partitionIdFunction)

Configure a Function that will be invoked at runtime to determine the partition id under which a message will be stored in the topic. Typically used with a Java 8 Lambda expression:

 .partitionId(m -> m.getHeaders().get("partitionId", Integer.class))
 
 

Type Parameters:

P - the expected payload type.

Parameters:

partitionIdFunction - the partitionId function.

Returns:

the spec.

partitionIdExpression

public S partitionIdExpression(Expression partitionIdExpression)

Configure an Expression to determine the topic partitionId at runtime against request Message as a root object of evaluation context.

Parameters:

partitionIdExpression - the partitionId expression to use.

Returns:

the spec.

timestampExpression

public S timestampExpression(String timestampExpression)

Configure a SpEL expression to determine the timestamp at runtime against a request Message as a root object of evaluation context.

Parameters:

timestampExpression - the timestamp expression to use.

Returns:

the spec.

timestamp

public <P> S timestamp(java.util.function.Function<Message<P>,Long> timestampFunction)

Configure a Function that will be invoked at runtime to determine the Kafka record timestamp will be stored in the topic. Typically used with a Java 8 Lambda expression:

 .timestamp(m -> m.getHeaders().get("mytimestamp_header", Long.class))
 
 

Type Parameters:

P - the expected payload type.

Parameters:

timestampFunction - the timestamp function.

Returns:

the spec.

timestampExpression

public S timestampExpression(Expression timestampExpression)

Configure an Expression to determine the timestamp at runtime against a request Message as a root object of evaluation context.

Parameters:

timestampExpression - the timestamp expression to use.

Returns:

the spec.

flushExpression

public S flushExpression(String flushExpression)

Configure a SpEL expression to determine whether or not to flush the producer after a send. By default the producer is flushed if a header kafka_flush has a value Boolean.TRUE.

Parameters:

flushExpression - the timestamp expression to use.

Returns:

the spec.

flush

public <P> S flush(java.util.function.Function<Message<P>,Boolean> flushFunction)

Configure a Function that will be invoked at runtime to determine whether or not to flush the producer after a send. By default the producer is flushed if a header kafka_flush has a value Boolean.TRUE. Typically used with a Java 8 Lambda expression:

 .flush(m -> m.getPayload().shouldFlush())
 
 

Type Parameters:

P - the expected payload type.

Parameters:

flushFunction - the flush function.

Returns:

the spec.

flushExpression

public S flushExpression(Expression flushExpression)

Configure an Expression to determine whether or not to flush the producer after a send. By default the producer is flushed if a header kafka_flush has a value Boolean.TRUE.

Parameters:

flushExpression - the timestamp expression to use.

Returns:

the spec.

sync

public S sync(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.

Returns:

the spec.

sendTimeout

public S sendTimeout(long sendTimeout)

Specify a timeout in milliseconds how long KafkaProducerMessageHandler should wait wait for send operation results. Defaults to 10 seconds.

Parameters:

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

Returns:

the spec.

headerMapper

public S headerMapper(org.springframework.kafka.support.KafkaHeaderMapper mapper)

Specify a header mapper to map spring messaging headers to Kafka headers.

Parameters:

mapper - the mapper.

Returns:

the spec.

sendSuccessChannel

public S sendSuccessChannel(MessageChannel sendSuccessChannel)

Set the channel to which successful send results are sent.

Parameters:

sendSuccessChannel - the channel.

Returns:

the spec.

Since:

3.0.2

sendSuccessChannel

public S sendSuccessChannel(String sendSuccessChannel)

Set the channel to which successful send results are sent.

Parameters:

sendSuccessChannel - the channel name.

Returns:

the spec.

Since:

3.0.2

sendFailureChannel

public S sendFailureChannel(MessageChannel sendFailureChannel)

Set the channel to which failed send results are sent.

Parameters:

sendFailureChannel - the channel.

Returns:

the spec.

Since:

3.0.2

sendFailureChannel

public S sendFailureChannel(String sendFailureChannel)

Set the channel to which failed send results are sent.

Parameters:

sendFailureChannel - the channel name.

Returns:

the spec.

Since:

3.0.2

futuresChannel

public S futuresChannel(MessageChannel futuresChannel)

Set the channel to which send futures are sent.

Parameters:

futuresChannel - the channel.

Returns:

the spec.

Since:

5.4

futuresChannel

public S futuresChannel(String futuresChannel)

Set the channel to which send futures are sent.

Parameters:

futuresChannel - the channel name.

Returns:

the spec.

Since:

5.4