Kafka 객체 Produce 및 Consume 중 발생한 에러

문제 발생

객체를 생성해서 카프카에 보내는 것은 잘되지만 객체를 받아오는 과정에서 문제가 발생하였다.
2023-05-14T16:48:02.653+09:00 ERROR 39835 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer : Consumer exception

ING

그래서 검색을 해서 찾아보니 직렬화와 역직렬화를 하는 과정에서

직렬화, 역직렬화 과정에서는 package 이름까지 포함하기때문에

consumer쪽의 JsonDeserializer에서 class is not in the trusted packages consumer쪽의 JsonDeserializer에서 class is not in the trusted packages 에러가 나오게된다.

그래서 이제 커스텀 JsonDeserializer를 만들어주려고
해당 글을 참고해서 https://sup2is.tistory.com/102 만든뒤

deserializer.setRemoveTypeHeaders(false);
deserializer.addTrustedPackages("*");
deserializer.setUseTypeMapperForKey(true);

실행을 해보았지만 이러한 에러가 다시 나를 반겼다.

org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry

엔드포인트.. 음...

문제 해결 완료

나는 한 프로젝트안에 여러 패키지를 구분하여 단계별로 String, Object 등 여러가지 Kafka Test를 진행하고 있었기에
YML에서 작성하는 것은 최대한 지양하고 있었다.

하지만 YML에서

spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=*

이렇게 작성하는 해결법(역직렬화를 정해주고 해당 옵션을 넣어주는 방향)
내가 찾아낸

@Bean
    public ConsumerFactory<String, StockChange> consumerFactory() {

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "foo");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); // 이부분
        return new DefaultKafkaConsumerFactory<>(props);
    }

를 사용해서 신뢰할수 있는 패키지를 구분해주는 것이 좋을 듯하다.
해당 "*"은 공부용 프로젝트라 전체 설정을 해주었지만 때에 따라 패키지 명을 명시해서 작성해주면 된다.

Reference

https://stackoverflow.com/questions/51688924/spring-kafka-the-class-is-not-in-the-trusted-packages