Is there a way to configure custom RecordMessageConverter in spring-kafka?

user_11960656 0
Viewed 967

I created a custom RecordMessageConverter which print in the toMessage and fromMessage (I create Bean as explained at https://www.confluent.io/blog/spring-for-apache-kafka-deep-dive-part-1-error-handling-message-conversion-transaction-support) but unfortunately it is not binded to the listener. Can I explicitly bind it or should I do something else to make it works?

public class MobileMessageConverter implements RecordMessageConverter {
    @Override
    public Message<?> toMessage(ConsumerRecord<?,?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer, Type payloadType) {
        System.out.println("Converting toMessage");
        return null;
    }

    @Override
    public ProducerRecord<CustomKey, CustomValue> fromMessage(Message<?> message, String defaultTopic) {
        System.out.println("Converting fromMessage");
        return new ProducerRecord("UNKNOWN",new CustomValue());
    }
}


@RestController
public class CustomKafkaController {
    @Bean
    public RecordMessageConverter converter() {
        return new CustomMessageConverter();
    }

    @KafkaListener(topics = "UNKNOWN", containerFactory = "kafkaListenerContainerFactory", groupId = "1")
    public void listenAsObject(ConsumerRecord<CustomKey, CustomValue> cr) {
        System.out.println(cr.getValue().getCustomMessage());
    }
}
spring-kafka apache-kafka spring-boot
1 Answers

I solve it!

  1. I added the converter to the factory Bean (kafkaListenerContainerFactory) by its setter.
  2. I changed the convert to implement MessagingMessageConverter insted of RecordMessageConverter and I implemented the method extractAndConvertValue
  3. (If you want to change to other type) remove the listenAsObject and replace it by the ConsumerRecordtype your convert to

For example:
If you want to change the custom value to String:
create new converter: 

public class CustomToStringMessageConverter extends MessagingMessageConverter {
    @Override
    protected String extractAndConvertValue(ConsumerRecord<?, ?> record, Type type){
       return "EmptyString"
    }
}

change the listener to : 

@KafkaListener(topics = "UNKNOWN", containerFactory = "kafkaListenerStringContainerFactory", groupId = "1")
public void listen2(String cr) {
    **DO SOMETHING WITH THE CONVERTED CR**
}

and the last thing is to get the converted Bean and set it in your factory by : 

factory.setMessageConverter(new CustomToStringMessageConverter ());
user_119606560
Related