CODEWITHSUNDEEP
javaspring-bootapache-kafkaspring-kafka
salvatore
salvatore

Reputation: 9

Why can I not set deserializer in consumer factory?

everybody.

package kitchen;

import entity.Order;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.converter.JsonMessageConverter;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@SpringBootApplication
public class PizzaKitchenApplication {
    public static void main(String[] args) {
        SpringApplication.run(PizzaKitchenApplication.class, args);
    }

    @Bean
    public ConsumerFactory<String, Order> consumerFactory(){
        Map<String, Object> configs = new HashMap<>();

        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "com.pizzapool");

        return new DefaultKafkaConsumerFactory<>(configs, new StringDeserializer(), new JsonDeserializer<>(Order.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Order> orderConcurrentKafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String, Order> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();

        containerFactory.setConsumerFactory(consumerFactory());
        containerFactory.setMessageConverter(new JsonMessageConverter());

        return containerFactory;
    }
}

It's my consumer config. When I'm running the application i see in console:

value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

But I set value deserializer to JsonDeserializer. I tried to set trough configuration map. In both cases it doesn't work. When I received the message I have:

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void kitchen.service.KafkaOrderMessagingService.handle(entity.Order) throws javax.jms.JMSException]
Bean [kitchen.service.KafkaOrderMessagingService@13183edf]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [entity.Order] for GenericMessage

Upvotes: 0

Views: 750

Answers (1)

planky
planky

Reputation: 440

As stated in the docs (https://docs.spring.io/spring-kafka/reference/html/#record-listener):

This mechanism requires an @EnableKafka annotation on one of your @Configuration classes and a listener container factory, which is used to configure the underlying ConcurrentMessageListenerContainer. By default, a bean with name kafkaListenerContainerFactory is expected.

Your bean is named orderConcurrentKafkaListenerContainerFactory.

You can use this default name or specify custom bean's name in the listener with

@KafkaListener(topics = "test", containerFactory = "orderConcurrentKafkaListenerContainerFactory")

The containerFactory is useful also when you need multiple different factories.

Upvotes: 1

Related Questions