When set ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to IntegerSerializer in ProducerConfigs in Spring boot kafka, it gives class cast exception

Question

I m using this Spring boot with kafka to setup my project. but when i run it it will give org.apache.kafka.common.errors.SerializationException: Can't convert key of class java.lang.Integer to class org.apache.kafka.common.serialization.StringSerializer specified in key.serializer Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.String this exception when send the request using kafka template.

@EmbeddedKafka(partitions = 1)
@SpringBootTest
class KafkaApplicationTests {

    @Autowired
    private MyKafkaListener listener;

    @Autowired
    private KafkaTemplate<Integer, String> template;

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Test
    public void testSimple() throws Exception {
        template.send("annotated1", 0, "foo");
        template.flush();
        assertTrue(this.listener.latch1.await(10, TimeUnit.SECONDS));
    }

    @Configuration
    @EnableKafka
    public class Config {

        @Bean
        ConcurrentKafkaListenerContainerFactory<Integer, String>
        kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                    new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            return factory;
        }

        @Bean
        public ConsumerFactory<Integer, String> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerConfigs());
        }

        @Bean
        public Map<String, Object> consumerConfigs() {
            Map<String, Object> props = new HashMap<>();
//          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            return props;
        }

        @Bean
        public ProducerFactory<Integer, String> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }

        @Bean
        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
//          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            return props;
        }

        @Bean
        public KafkaTemplate<Integer, String> kafkaTemplate() {
            return new KafkaTemplate<Integer, String>(producerFactory());
        }

    }
}

But when I change that template.send("annotated1", "key-foo", "foo"); it will work. I have used

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    return props;
}

this configuration as well. But it is still giving me the Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.String

If someone there please help me. thanks

Answer 1

You need to add the main application class and Config class to the @SpringBootTest

@SpringBootTest(classes = { So61985794Application.class, So61985794ApplicationTests.Config.class })

to override Boot's normal configuration.