[Spring Boot] Spring Kafka Consumer에서 JSON 처리하기

 

 

참고 https://huigrowthdiary.tistory.com/88

[Spring Boot] Kafka 기반 주문 서비스와 상품 서비스의 비동기 재고 차감

 

위 포스팅에서는 카프카 기반으로 주문 생성 시 상품 재고 업데이트를 하는 것을 작성했습니다.

문제점은 Kafka로 넘어오는 Message를 자바 객체로 변환하려고하고, 파싱할 때 Map<> 객체로 직접 받아와서 타입이 안전하지 않아 보였고, 이로 인해 발생할 문제를 사전에 차단하기 위해 방법을 생각하였습니다. 

 

우선 그 방법을 작성하기 전에 카프카 통신이랑 조금 더 친해지기 위해 위 포스팅 내용에 이어지는 것으로 정리 하나만 하고 가겠습니다.

 

 

@KafkaListener

---

 

@KafkaListener는 Spring Kafka에서 제공하는 어노테이션으로, Consumer가 특정 Topic을 자동으로 구독하도록 만드는 기능을 합니다. 

 

@KafkaListener(topics = "example-catalog-topic")
public void updateQty(String kafkaMessage) {
    ...
}

 

- Listener Container가 topic을 지속적으로 polling 합니다. 

- 새로운 메시지가 들어오면 updateQty() 메서드를 자동 호출합니다. 

- KafkaMessage에는 "Product가 보낸 JSON문자열"이 그대로 들어옵니다. 

 

따라서 Consumer에서는 Producer와 다르게 KafkaTemplate을 호출할 필요가 없습니다. 

KafkaTemplate은 Producer가 메시지를 보낼 때만 사용하는 객체입니다. 

 

 

Map으로 처리하는 방식 

---

 

이전 포스팅에서, 그리고 소스코드를 업데이트하기 전에는 JSON -> Map으로 역직렬화하여 사용하는 방법이었습니다. 

 

@KafkaListener(topics = "example-catalog-topic")
public void updateQty(String kafkaMessage) {

    log.info("Kafka Message : {}", kafkaMessage);

    Map<Object, Object> map = new HashMap<>();
    ObjectMapper mapper = new ObjectMapper();

    try {
        // JSON 문자열을 Map 형태로 변환
        map = mapper.readValue(
                kafkaMessage,
                new TypeReference<Map<Object, Object>>() {}
        );
    } catch (JsonProcessingException e) {
        e.printStackTrace();
    }

    CatalogEntity entity = catalogJpaRepository.findByProductId((String)map.get("productId"))
            .orElseThrow(() -> new RuntimeException("product not found"));

    entity.updateStock((Integer) map.get("qty"));
    catalogJpaRepository.save(entity);
}

 

여기서 TypeReperence의 역할은 

Jackson은 제네릭 타입(Map, List, Dto 리스트 등) 역직렬화 시 타입 정보를 정확히 알지 못합니다.

이것을 해결하기 위해 TypeReperence가 필요합니다. 

 

new TypeReference<Map<Object, Object>>() {}

 

이는 mapper가 "Map<Object, Object> 타입으로 변환해야 한다"는 정보를 제공합니다. 

 

 

Map 방식의 문제점

---

 

 

문제점	설명
타입 안전성 없음	"qty" 오타 나면 Null 리턴
캐스팅 필요	Object → Integer 타입 캐스팅 필요
유지보수 어려움	필드명 변경 시 오류가 컴파일에서 잡히지 않음
JSON 구조가 복잡해지면 확장 어려움	실무에서 payload가 커질수록 Map 방식은 비추천

 

 

DTO로 받아서 처리하는 개선된 방법

---

 

DTO생성

@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class OrderMessage {

    private String productId;
    private Integer qty;
    private Integer unitPrice;

}

 

Consumer에서 DTO로 변환하여 처리

@KafkaListener(topics = "example-catalog-topic")
public void updateQty(String kafkaMessage) {

    log.info("Kafka Message : {}", kafkaMessage);

    ObjectMapper mapper = new ObjectMapper();
    OrderMessage orderMessage = null;

    try {
        // JSON → DTO 변환
        orderMessage = mapper.readValue(kafkaMessage, OrderMessage.class);
    } catch (JsonProcessingException e) {
        e.printStackTrace();
    }

    CatalogEntity entity = catalogJpaRepository
            .findByProductId(orderMessage.getProductId())
            .orElseThrow(() -> new RuntimeException("product not found"));

    entity.updateStock(orderMessage.getQty());
    catalogJpaRepository.save(entity);
}

 

 

DTO 방식의 장점

---

 

 

장점	설명
타입 안전성	productId, qty 오타 → 컴파일 에러로 잡힘
캐스팅 필요 없음	더 이상 map.get() 사용 X
유지보수 용이	JSON 구조가 바뀌면 DTO만 수정하면 됨
가독성 향상	어떤 데이터가 오고가는지 명확함

 

 

번외(엔티티로 직접 받으면 안되는 이유)

---

 

Kafka 메시지는 "DB 상태"가 아니라 "이벤트/명령" 데이터이기 때문에 JPA엔티티로 직접 받으면 안된다고 합니다. 

 

엔티티는 연관관계/ 프록시/ 영속성 컨텍스트 상태를 포함합니다.

JSON으로 표현할 수 없는 JPA 내부 정보가 많습니다. 

엔티티는 "DB상태를 표현하는 객체"이지 "메시지 전달 객체"가 아닙니다.

엔티티 구조 변경되면 메시지 형식까지 깨지기 때문에 강결합이 발생합니다. 

 

따라서 엔티티로 직접 받는 것은 잘못된 설계라고 생각이 들어 귀찮더라도 DTO로 받는 것이 정답입니다. 

 

 

정리

---

 

처음에는 Json -> Map으로 받는 것이 편하다고 생각을 했고 실제로 Map으로 받으면 편하긴 했다. 

필드를 추가하거나 구조가 바뀌어도 그냥 Key만 꺼내서 사용하면 되기 때문이었다. 

 

그런데, 타입이 Object라서 계속 캐스팅을 강제로 해줘야 했다. 

그리고, key가 오타 나도 컴파일 오류 시점에 잡히지 않는다. 이 말은 런타임 에러시점에 걸린다는 것이다. 

 

추가로 Map을 사용할 때 TypeReperence를 사용해야 했는데, 이해도 하려고 하니까 Map 방식은 원래 복잡한 방식이라는 것을 깨달았다. Map으로 변환하려면 반드시 TypeReperence를 사용해야 한다. 

이유는 Jackson이 제네릭 타입의 실제 정보를 알 수 없기 때문이다. 

여기서 느낀 것은 Jackson은 그냥 Map으로 때려 넣는 걸 좋아하지 않는 다는 것을 느꼈다. 

편해 보이지만 번거롭고 안정성이 떨어지는 방식이라는 것을 알게되었다. 

 

그래서 DTO를 도입했더니 캐스팅이 따로 필요없고, 필드명이 틀리면 컴파일에서 바로 잡아준다. 

그리고 코드를 보면 알겠지만 가독성이 좋아졌다.