Consume XML from Kafka topic

I need to consume XML type from Kafka topic. After reading through few resources I came to conclusion I might have to write my own "serializer". However I am totally lost on

• How to write such serializer (for value) in first place? and
• How to inject this serializer in Kafka configuration (not just assign class name in spring.deserializer.value.deegate.class)

With raw Kafka client (org.apache.kafka.kafka-clients) it's pretty simple:

1. Implement an org.apache.kafka.common.serialization.Deserializer interface. Here is a tutorial. In a nutshell, you have to implement single method that turns byte[] into object of any class you like. Do all xml stuff here and return a parsed xml. Lets say the implementation is named XmlDeserializer.
2. Use KafkaConsumer constructor with a 3-arg signature and pass a deserializer into it: new KafkaConsumer(props, keyDeserializer, new XmlDeserializer()). Use keyDeserializer you need, there are plenty predifined of them, for example, new StringDeserializer()

That's all. You will get a KafkaConsumer typed with classes representing parsed key and value. Every poll() would return objects of that classes. Your XmlDeserializer would be called automatically under the hood.

If using Spring Kafka, you probably create DefaultKafkaConsumerFactory as a bean. In step 2) pass new XmlDeserializer() as a constructor parameter.

Also, instead of passing new XmlDeserializer() to a constructor (either KafkaConsumer or DefaultKafkaConsumerFactory) you could put deserializer's class name into props: props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, XmlDeserializer.class) .

How do you currently read XML data that isn't a file?

Just use that, and configure your consumer with ByteArrayDeserializer or StringDeserializer.

The only thing a custom class would do is wrap the same functions you'd use with the other two.