No String-argument constructor/factory method to deserialize from String value

개요

• 문제 상황
• 해결 방법
• 설정 방법

문제 상황

Kafka에서 Json 형식의 데이터를 송수신하는 과정에서 데이터를 보낼 수 없다는 에러가 발생

해결 방법

Kafka를 사용하여 json의 데이터를 보내는 경우에는 직렬화 형식을 json으로 수정하여야 합니다. ❗중요한 점은 consume과정에서 ErrorHandlingDeserializer를 해주어야 합니다.

ErrorHandlingDeserializer을 사용하는 이유

1. consume과정에서 사용자가 정의한 ErrorHandler는 작동하지 않습니다.
2. 데이터의 형식이 잘못 들어오면 Kafka가 consume을 무한히 반복합니다.

• consume과정에서 사용자가 정의한 ErrorHandler는 작동하지 않는 이유

kafka의 consume 과정은 아래 4단계를 따릅니다.

1. 브로커로부터 직렬화된 데이터를 가져옴
2. 데이터 역직렬화
3. 역질렬화한 데이터를 처리 (사용자가 작성한 처리 로직)
4. commit요청을 브로커에서 보냄

사용자가 선언한 Handler는 c 단계에 있지만, consume 과정에서 발생하는 에러는 b 단계에서 발생합니다.

따라서, b 단계의 에러는 사용자가 처리하기 힘듭니다.

이를 해결하기 위해 Kafka는 b단계에서의 에러를 처리해주는 ErrorHandlingDeserializer을 도입하였습니다.

ErrorHandlingDeserializer를 사용하면 b 단계에서 에러가 발생하여도, 결과값으로 null을 넣어주어 c 단계까지 데이터를 넘겨주게 되어 사용자가 에러를 처리할 수 있게 됩니다.

설정 방법

저는 기본적인 props들은 application.properties에 있는 설정을 사용하여 새로운 Factory를 생성하였습니다.

• producer properties에 아래 내용을 추가합니다.

@Bean
public ProducerFactory<String, Object> producerJsonFactory() {
    Map<String, Object> props  = new HashMap<>();
    ...
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
		...
    return new DefaultKafkaProducerFactory<>(props);
}

• consumer properties에 아래 내용을 추가해줍니다.

@Bean
public ConsumerFactory<String, Object> batchConsumerFactory() {
	Map<String, Object> props = new HashMap<>();
	...
	props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
	props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
	
	props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
	props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
	props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
  ...
	return new DefaultKafkaConsumerFactory<>(props);
}

위 설정들을 추가한 consumer을 사용하면 json 형식의 데이터를 전달받을 수 있습니다.

 

REF: https://velog.io/@kshired/Spring-Kafka-ErrorHandlingDeserializer

https://junuuu.tistory.com/793

https://velog.io/@yanggi0824/Spring-Boot-Kafka-Json-전송하기