스프링 kafka 역직렬화 문제

Question

스프링에 카프카를 적용하고 토픽에 수동으로 메시지를 보내는 중 발생한 이슈입니다.java.lang.IllegalStateException: This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer at org.springframework.kafka.listener.DefaultErrorHandler.handleOtherException(DefaultErrorHandler.java:192) ~[spring-kafka-3.1.1.jar:3.1.1] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1934) ~[spring-kafka-3.1.1.jar:3.1.1] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1365) ~[spring-kafka-3.1.1.jar:3.1.1] at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804) ~[na:na] at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na] Caused by: org.apache.kafka.common.errors.RecordDeserializationException: Error deserializing key/value for partition like-0 at offset 1. If needed, please seek past the record to continue consumption. at org.apache.kafka.clients.consumer.internals.CompletedFetch.parseRecord(CompletedFetch.java:309) ~[kafka-clients-3.6.1.jar:na] at org.apache.kafka.clients.consumer.internals.CompletedFetch.fetchRecords(CompletedFetch.java:263) ~[kafka-clients-3.6.1.jar:na] at org.apache.kafka.clients.consumer.internals.AbstractFetch.fetchRecords(AbstractFetch.java:340) ~[kafka-clients-3.6.1.jar:na] at org.apache.kafka.clients.consumer.internals.AbstractFetch.collectFetch(AbstractFetch.java:306) ~[kafka-clients-3.6.1.jar:na] at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1235) ~[kafka-clients-3.6.1.jar:na] at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1186) ~[kafka-clients-3.6.1.jar:na] at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1159) ~[kafka-clients-3.6.1.jar:na] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollConsumer(KafkaMessageListenerContainer.java:1649) ~[spring-kafka-3.1.1.jar:3.1.1] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1624) ~[spring-kafka-3.1.1.jar:3.1.1] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1421) ~[spring-kafka-3.1.1.jar:3.1.1] at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1313) ~[spring-kafka-3.1.1.jar:3.1.1] ... 2 common frames omitted Caused by: org.apache.kafka.common.errors.SerializationException: Can't deserialize data from topic [like] at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:588) ~[spring-kafka-3.1.1.jar:3.1.1] at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:73) ~[kafka-clients-3.6.1.jar:na] at org.apache.kafka.clients.consumer.internals.CompletedFetch.parseRecord(CompletedFetch.java:300) ~[kafka-clients-3.6.1.jar:na] ... 12 common frames omitted Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'asd': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (byte[])"asd"; line: 1, column: 4] at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:2477) ~[jackson-core-2.15.3.jar:2.15.3] at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:760) ~[jackson-core-2.15.3.jar:2.15.3] at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3699) ~[jackson-core-2.15.3.jar:2.15.3] at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2787) ~[jackson-core-2.15.3.jar:2.15.3] at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:908) ~[jackson-core-2.15.3.jar:2.15.3] at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:794) ~[jackson-core-2.15.3.jar:2.15.3] at com.fasterxml.jackson.databind.ObjectReader._initForReading(ObjectReader.java:357) ~[jackson-databind-2.15.3.jar:2.15.3] at com.fasterxml.jackson.databind.ObjectReader._bindAndClose(ObjectReader.java:2095) ~[jackson-databind-2.15.3.jar:2.15.3] at com.fasterxml.jackson.databind.ObjectReader.readValue(ObjectReader.java:1583) ~[jackson-databind-2.15.3.jar:2.15.3] at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:585) ~[spring-kafka-3.1.1.jar:3.1.1] ... 14 common frames omittedjson 형식으로는 잘 받아지는데일부러 잘못된 형식(asd)으로 보냈더니 이런 예외 로그가 무한 반복됩니다.ErrorHandlingDeserializer를 추가해봤지만 여전히 같은 문제가 발생하고 있습니다.(yml로 설정하는 방법도 해봤습니다.)@Bean public Map consumerConfig() { Map props = new HashMap<>(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER); props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-01"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class); props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName()); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); return props; }해결 방법 있을까요?참고 자료https://docs.spring.io/spring-kafka/reference/kafka/serdes.htmlhttps://serverwizard.tistory.com/285

kkey21a · Answer

에러나도 commit 처리 하면 안되는 부분일까요?

컴포지트 · Answer

이 글과 유사한 듯 한데 참고해 보시길https://velog.io/@mouse4786/%EC%B9%B4%ED%94%84%EC%B9%B4-error.tolerance%EC%97%90-%EB%8C%80%ED%95%B4

스프링 kafka 역직렬화 문제

2개의 답변

Q&A 목록