org.axonframework.kafka.eventhandling

Class DefaultKafkaMessageConverter

java.lang.Object

org.axonframework.kafka.eventhandling.DefaultKafkaMessageConverter

All Implemented Interfaces:

KafkaMessageConverter<String,byte[]>

public class DefaultKafkaMessageConverter
extends Object
implements KafkaMessageConverter<String,byte[]>

Converts: EventMessage to kafkaMessage and message read from Kafka back to EventMessage (if possible).

During conversion it passes all meta-data entries with 'axon-metadata-' prefix to Headers. Other message-specific attributes are added as metadata. The payload is serialized using the configured Serializer and passed as the message body.

This implementation will suffice in most cases.

Since:

3.0

Author:

Nakul Mishra

Constructor Summary

Constructors

Constructor and Description
DefaultKafkaMessageConverter(Serializer serializer) Initializes the KafkaMessageConverter with the given serializer.
DefaultKafkaMessageConverter(Serializer serializer, SequencingPolicy<? super EventMessage<?>> sequencingPolicy, BiFunction<String,Object,org.apache.kafka.common.header.internals.RecordHeader> headerValueMapper) Initializes the KafkaMessageConverter with the given serializer, sequencingPolicy and objectMapper.

Method Summary

Modifier and Type	Method and Description
org.apache.kafka.clients.producer.ProducerRecord<String,byte[]>	createKafkaMessage(EventMessage<?> eventMessage, String topic) Creates ProducerRecord for a given EventMessage
Optional<EventMessage<?>>	readKafkaMessage(org.apache.kafka.clients.consumer.ConsumerRecord<String,byte[]> consumerRecord) Reconstruct an EventMessage from the given ConsumerRecord.

Methods inherited from class java.lang.Object

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

Constructor Detail

DefaultKafkaMessageConverter

public DefaultKafkaMessageConverter(Serializer serializer)

Initializes the KafkaMessageConverter with the given serializer.

Parameters:

serializer - The serializer to serialize the Event Message's payload with.

DefaultKafkaMessageConverter

public DefaultKafkaMessageConverter(Serializer serializer,
                                    SequencingPolicy<? super EventMessage<?>> sequencingPolicy,
                                    BiFunction<String,Object,org.apache.kafka.common.header.internals.RecordHeader> headerValueMapper)

Initializes the KafkaMessageConverter with the given serializer, sequencingPolicy and objectMapper.

Parameters:

serializer - The serializer to serialize the Event Message's payload and Meta Data with

sequencingPolicy - The policy to generate the key of the ProducerRecord.

headerValueMapper - The Function for mapping values to Kafka headers.

Method Detail

createKafkaMessage

public org.apache.kafka.clients.producer.ProducerRecord<String,byte[]> createKafkaMessage(EventMessage<?> eventMessage,
                                                                                          String topic)

Description copied from interface: KafkaMessageConverter

Creates ProducerRecord for a given EventMessage

Specified by:

createKafkaMessage in interface KafkaMessageConverter<String,byte[]>

Parameters:

eventMessage - the event message to send to Kafka.

topic - the Kafka topic.

Returns:

the record.

readKafkaMessage

public Optional<EventMessage<?>> readKafkaMessage(org.apache.kafka.clients.consumer.ConsumerRecord<String,byte[]> consumerRecord)

Description copied from interface: KafkaMessageConverter

Reconstruct an EventMessage from the given ConsumerRecord. The returned optional resolves to a message if the given input parameters represented a correct EventMessage.

Specified by:

readKafkaMessage in interface KafkaMessageConverter<String,byte[]>

Parameters:

consumerRecord - Event message represented inside kafka

Returns:

The Event Message to publish on the local event processors