org.springframework.kafka.support.converter

Class MessagingMessageConverter

java.lang.Object

org.springframework.kafka.support.converter.MessagingMessageConverter

All Implemented Interfaces:

MessageConverter, RecordMessageConverter

Direct Known Subclasses:

JsonMessageConverter, ProjectingMessageConverter

public class MessagingMessageConverter
extends java.lang.Object
implements RecordMessageConverter

A Messaging MessageConverter implementation for a message listener that receives individual messages.

Populates KafkaHeaders based on the ConsumerRecord onto the returned message.

Author:

Marius Bogoevici, Gary Russell, Dariusz Szablinski, Biju Kunjummen

Field Summary

Fields

Modifier and Type	Field and Description
protected org.springframework.core.log.LogAccessor	logger

Constructor Summary

Constructors

Constructor and Description
MessagingMessageConverter()

Method Summary

Modifier and Type	Method and Description
protected java.lang.Object	convertPayload(org.springframework.messaging.Message<?> message) Subclasses can convert the payload; by default, it's sent unchanged to Kafka.
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.
org.apache.kafka.clients.producer.ProducerRecord<?,?>	fromMessage(org.springframework.messaging.Message<?> message, java.lang.String defaultTopic) Convert a message to a producer record.
protected org.apache.kafka.common.header.Headers	initialRecordHeaders(org.springframework.messaging.Message<?> message) Subclasses can populate additional headers before they are mapped.
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(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record, Acknowledgment acknowledgment, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, java.lang.reflect.Type type) Convert a ConsumerRecord to a Message.

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.support.converter.MessageConverter

commonHeaders, getGroupId

Field Detail

logger

protected final org.springframework.core.log.LogAccessor logger

Constructor Detail

MessagingMessageConverter

public MessagingMessageConverter()

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(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record,
                                                          Acknowledgment acknowledgment,
                                                          org.apache.kafka.clients.consumer.Consumer<?,?> consumer,
                                                          java.lang.reflect.Type type)

Description copied from interface: RecordMessageConverter

Convert a ConsumerRecord to a Message.

Specified by:

toMessage in interface RecordMessageConverter

Parameters:

record - the record.

acknowledgment - the acknowledgment.

consumer - the consumer

type - the required payload type.

Returns:

the message.

fromMessage

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

Description copied from interface: RecordMessageConverter

Convert a message to a producer record.

Specified by:

fromMessage in interface RecordMessageConverter

Parameters:

message - the message.

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

Returns:

the producer record.

initialRecordHeaders

protected org.apache.kafka.common.header.Headers initialRecordHeaders(org.springframework.messaging.Message<?> message)

Subclasses can populate additional headers before they are mapped.

Parameters:

message - the message.

Returns:

the headers

Since:

2.1

convertPayload

protected java.lang.Object convertPayload(org.springframework.messaging.Message<?> message)

Subclasses can convert the payload; by default, it's sent unchanged to Kafka.

Parameters:

message - the message.

Returns:

the payload.

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.

Parameters:

record - the record.

type - the required type.

Returns:

the value.