🐛 fix: Kafka Consumer Type Mapping 설정 추가

WaitherTeam · Jul 5, 2024 · 70c7425 · 70c7425
2 parents 4c4d3c9 + 286f4af
commit 70c7425
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 5 deletions.
diff --git a/noti-service/src/main/java/com/waither/notiservice/config/KafkaConsumerConfig.java b/noti-service/src/main/java/com/waither/notiservice/config/KafkaConsumerConfig.java
@@ -137,8 +137,9 @@ public ConcurrentKafkaListenerContainerFactory<String, KafkaDto.WeatherDto> weat
 
     private ConsumerFactory<String, KafkaDto.WeatherDto> weatherConsumerFactory() {
         Map<String, Object> props = dtoSettings();
-        props.put(JsonDeserializer.TYPE_MAPPINGS, "KafkaMessage:com.waither.notiservice.dto.kafka.KafkaDto.WeatherDto");
-        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.waither.*");
+        props.put(JsonDeserializer.TYPE_MAPPINGS, "com.waither.weatherservice.kafka.KafkaMessage:com.waither.notiservice.dto.kafka.KafkaDto$WeatherDto");
+        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.waither.*,com.waither.weatherservice.kafka");
+        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.waither.notiservice.dto.kafka.KafkaDto$WeatherDto");
         return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), createJsonDeserializer(KafkaDto.WeatherDto.class));
     }
 

diff --git a/noti-service/src/main/java/com/waither/notiservice/service/KafkaConsumer.java b/noti-service/src/main/java/com/waither/notiservice/service/KafkaConsumer.java
@@ -13,6 +13,7 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -108,7 +109,7 @@ public void consumeUserInit(KafkaDto.InitialDataDto initialDataDto) {
      * @Query  : 0200, 0500, 0800, 1100, 1400, 1700, 2000, 2300  */
     @Transactional
     @KafkaListener(topics = "alarm-wind", containerFactory = "weatherKafkaListenerContainerFactory")
-    public void consumeWindAlarm(KafkaDto.WeatherDto weatherDto) {
+    public void consumeWindAlarm(@Payload KafkaDto.WeatherDto weatherDto) {
 
         int currentHour = LocalDateTime.now().getHour();
         // 22:00 ~ 07:00 는 알림을 전송하지 않음
@@ -165,7 +166,7 @@ public void consumeWindAlarm(KafkaDto.WeatherDto weatherDto) {
      */
     @Transactional
     @KafkaListener(topics = "alarm-rain", containerFactory = "weatherKafkaListenerContainerFactory")
-    public void consumeRain(KafkaDto.WeatherDto weatherDto) {
+    public void consumeRain(@Payload KafkaDto.WeatherDto weatherDto) {
 
         int currentHour = LocalDateTime.now().getHour();
         // 22:00 ~ 07:00 는 알림을 전송하지 않음
@@ -230,7 +231,7 @@ public void consumeRain(KafkaDto.WeatherDto weatherDto) {
      * */
     @Transactional
     @KafkaListener(topics = "alarm-climate", containerFactory = "weatherKafkaListenerContainerFactory")
-    public void consumeClimateAlarm(KafkaDto.WeatherDto weatherDto) {
+    public void consumeClimateAlarm(@Payload KafkaDto.WeatherDto weatherDto) {
         int currentHour = LocalDateTime.now().getHour();
         // 22:00 ~ 07:00 는 알림을 전송하지 않음
         if (currentHour >= 22 || currentHour <= 7) {