feat: Kafka 리스너에 허용된 토픽 검증 로직 추가

Author: rnqhstmd

apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java

@@ -11,11 +11,13 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
+import java.util.Set;
 
 @Slf4j
 @Component
@@ -25,6 +27,15 @@ public class ProductMetricsConsumer {
     private final ProductMetricsFacade productMetricsFacade;
     private final ObjectMapper objectMapper;
 
+    @Value("${kafka.topic.product-like-name}")
+    private String productLikeTopic;
+
+    @Value("${kafka.topic.product-stock-name}")
+    private String productStockTopic;
+
+    @Value("${kafka.topic.product-view-name}")
+    private String productViewTopic;
+
     @KafkaListener(
             topics = {"${kafka.topic.product-like-name}", "${kafka.topic.product-stock-name}", "${kafka.topic.product-view-name}"},
             groupId = "${kafka.consumer.product-metrics-group}",
@@ -58,6 +69,13 @@ public void listen(
     }
 
     private void processPayload(String topic, String payload) throws JsonProcessingException {
+        Set<String> allowedTopics = Set.of(productLikeTopic, productStockTopic, productViewTopic);
+
+        if (!allowedTopics.contains(topic)) {
+            log.warn("허용되지 않은 토픽: {}", topic);
+            return;
+        }
+
         if (topic.contains("product-like")) {
             ProductLikePayload likePayload = objectMapper.readValue(payload, ProductLikePayload.class);
             ProductMetricsCommand likeCommand = ProductMetricsCommand.from(likePayload);