org.springframework.kafka.support.converter

Class BatchMessagingMessageConverter

java.lang.Object

org.springframework.kafka.support.converter.BatchMessagingMessageConverter

All Implemented Interfaces:

BatchMessageConverter, MessageConverter

public class BatchMessagingMessageConverter
extends java.lang.Object
implements BatchMessageConverter

A Messaging MessageConverter implementation used with a batch message listener; the consumer record values are extracted into a collection in the message payload.

Populates KafkaHeaders based on the ConsumerRecord onto the returned message. Each header is a collection where the position in the collection matches the payload position.

If a RecordMessageConverter is provided, and the batch type is a ParameterizedType with a single generic type parameter, each record will be passed to the converter, thus supporting a method signature List<Foo> foos.

Since:

1.1

Author:

Marius Bogoevici, Gary Russell, Dariusz Szablinski, Biju Kunjummen

Field Summary

Fields

Modifier and Type	Field and Description
protected org.apache.commons.logging.Log	logger

Constructor Summary

Constructors

Constructor and Description
BatchMessagingMessageConverter() Create an instance that does not convert the record values.
BatchMessagingMessageConverter(RecordMessageConverter recordConverter) Create an instance that converts record values using the supplied converter.

Method Summary

Modifier and Type	Method and Description
protected java.lang.Object	convert(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record, java.lang.reflect.Type type) Convert the record value.
protected java.lang.Object	extractAndConvertValue(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record, java.lang.reflect.Type type) Subclasses can convert the value; by default, it's returned as provided by Kafka unless a RecordMessageConverter has been provided.
java.util.List<org.apache.kafka.clients.producer.ProducerRecord<?,?>>	fromMessage(org.springframework.messaging.Message<?> message, java.lang.String defaultTopic) Convert a message to a producer record.
void	setGenerateMessageId(boolean generateMessageId) Generate Message ids for produced messages.
void	setGenerateTimestamp(boolean generateTimestamp) Generate timestamp for produced messages.
void	setHeaderMapper(KafkaHeaderMapper headerMapper) Set the header mapper to map headers.
org.springframework.messaging.Message<?>	toMessage(java.util.List<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>> records, Acknowledgment acknowledgment, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, java.lang.reflect.Type type) Convert a list of ConsumerRecord to a Message.

Methods inherited from class java.lang.Object

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

Field Detail

logger

protected final org.apache.commons.logging.Log logger

Constructor Detail

BatchMessagingMessageConverter

public BatchMessagingMessageConverter()

Create an instance that does not convert the record values.

BatchMessagingMessageConverter

public BatchMessagingMessageConverter(RecordMessageConverter recordConverter)

Create an instance that converts record values using the supplied converter.

Parameters:

recordConverter - the converter.

Since:

1.3.2

Method Detail

setGenerateMessageId

public void setGenerateMessageId(boolean generateMessageId)

Generate Message ids for produced messages. If set to false, will try to use a default value. By default set to false.

Parameters:

generateMessageId - true if a message id should be generated

setGenerateTimestamp

public void setGenerateTimestamp(boolean generateTimestamp)

Generate timestamp for produced messages. If set to false, -1 is used instead. By default set to false.

Parameters:

generateTimestamp - true if a timestamp should be generated

setHeaderMapper

public void setHeaderMapper(KafkaHeaderMapper headerMapper)

Set the header mapper to map headers.

Parameters:

headerMapper - the mapper.

Since:

1.3

toMessage

public org.springframework.messaging.Message<?> toMessage(java.util.List<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>> records,
                                                          Acknowledgment acknowledgment,
                                                          org.apache.kafka.clients.consumer.Consumer<?,?> consumer,
                                                          java.lang.reflect.Type type)

Description copied from interface: BatchMessageConverter

Convert a list of ConsumerRecord to a Message.

Specified by:

toMessage in interface BatchMessageConverter

Parameters:

records - the records.

acknowledgment - the acknowledgment.

consumer - the consumer.

type - the required payload type.

Returns:

the message.

fromMessage

public java.util.List<org.apache.kafka.clients.producer.ProducerRecord<?,?>> fromMessage(org.springframework.messaging.Message<?> message,
                                                                                         java.lang.String defaultTopic)

Description copied from interface: BatchMessageConverter

Convert a message to a producer record.

Specified by:

fromMessage in interface BatchMessageConverter

Parameters:

message - the message.

defaultTopic - the default topic to use if no header found.

Returns:

the producer records.

extractAndConvertValue

protected java.lang.Object extractAndConvertValue(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record,
                                                  java.lang.reflect.Type type)

Subclasses can convert the value; by default, it's returned as provided by Kafka unless a RecordMessageConverter has been provided.

Parameters:

record - the record.

type - the required type.

Returns:

the value.

convert

protected java.lang.Object convert(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record,
                                   java.lang.reflect.Type type)

Convert the record value.

Parameters:

record - the record.

type - the type - must be a ParameterizedType with a single generic type parameter.

Returns:

the converted payload.