refactor: RetryTracker 제거 및 Kafka 기본 재시도 메커니즘 활용

- ConcurrentHashMap 기반 수동 재시도 카운터 제거 (메모리 누수 위험 해소)
- DlqMessage retryCount 필드 제거
- DlqPublisher.publishToDlq() retryCount 파라미터 제거
- ProductMetricsConsumer에서 RetryTracker 의존성 제거
- 에러 처리 단순화: 복구 불가능한 에러는 즉시 DLQ, 일시적 에러는 Kafka 자동 재시도
- 관련 테스트 코드 업데이트

Author: sylee6529

apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessage.java

@@ -80,12 +80,6 @@ public class DlqMessage {
     @Column(name = "failed_at", nullable = false)
     private ZonedDateTime failedAt;
 
-    /**
-     * 재처리 시도 횟수
-     */
-    @Column(name = "retry_count", nullable = false)
-    private int retryCount = 0;
-
     /**
      * 재처리 완료 여부
      */
@@ -101,8 +95,7 @@ public static DlqMessage create(
             Long originalOffset,
             String messageKey,
             String messageValue,
-            Exception exception,
-            int retryCount
+            Exception exception
     ) {
         DlqMessage dlqMessage = new DlqMessage();
         dlqMessage.originalTopic = originalTopic;
@@ -114,7 +107,6 @@ public static DlqMessage create(
         dlqMessage.errorMessage = exception.getMessage();
         dlqMessage.stackTrace = getStackTraceAsString(exception);
         dlqMessage.failedAt = ZonedDateTime.now();
-        dlqMessage.retryCount = retryCount;
         dlqMessage.resolved = false;
         return dlqMessage;
     }

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/DlqPublisher.java

@@ -25,19 +25,17 @@ public class DlqPublisher {
      *
      * @param record 실패한 Kafka 메시지
      * @param exception 발생한 예외
-     * @param retryCount 재시도 횟수
      */
     @Transactional
-    public void publishToDlq(ConsumerRecord<String, String> record, Exception exception, int retryCount) {
+    public void publishToDlq(ConsumerRecord<String, String> record, Exception exception) {
         try {
             DlqMessage dlqMessage = DlqMessage.create(
                 record.topic(),
                 record.partition(),
                 record.offset(),
                 record.key(),
                 record.value(),
-                exception,
-                retryCount
+                exception
             );
 
             dlqMessageRepository.save(dlqMessage);

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/RetryTracker.java

@@ -9,7 +9,6 @@
 import com.loopers.application.metrics.MetricsAggregationService;
 import com.loopers.confg.kafka.KafkaConfig;
 import com.loopers.infrastructure.kafka.DlqPublisher;
-import com.loopers.infrastructure.kafka.RetryTracker;
 import com.loopers.infrastructure.kafka.event.KafkaEventEnvelope;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -26,6 +25,7 @@
  * - Product 관련 이벤트를 수신하여 메트릭 집계
  * - Manual Ack: 처리 성공 후에만 offset commit
  * - Batch Listener: 한 번에 여러 메시지 처리
+ * - 에러 처리: 복구 불가능한 에러는 DLQ로 전송, 일시적 에러는 Kafka 재시도
  */
 @Slf4j
 @Component
@@ -35,7 +35,6 @@ public class ProductMetricsConsumer {
     private final MetricsAggregationService aggregationService;
     private final ObjectMapper objectMapper;
     private final DlqPublisher dlqPublisher;
-    private final RetryTracker retryTracker;
 
     /**
      * 상품 좋아요 이벤트 Consumer
@@ -66,9 +65,6 @@ public void consumeProductLiked(
                     envelope.payload()
                 );
 
-                // 성공 시 재시도 카운터 제거
-                retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
-
             } catch (Exception e) {
                 log.error("[Consumer] Failed to process product-liked event - offset: {}, key: {}",
                     record.offset(), record.key(), e);
@@ -117,8 +113,6 @@ public void consumeProductUnliked(
                     envelope.payload()
                 );
 
-                retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
-
             } catch (Exception e) {
                 log.error("[Consumer] Failed to process product-unliked event - offset: {}, key: {}",
                     record.offset(), record.key(), e);
@@ -166,8 +160,6 @@ public void consumeOrderCompleted(
                     envelope.payload()
                 );
 
-                retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
-
             } catch (Exception e) {
                 log.error("[Consumer] Failed to process order-completed event - offset: {}, key: {}",
                     record.offset(), record.key(), e);
@@ -215,8 +207,6 @@ public void consumeProductViewed(
                     envelope.payload()
                 );
 
-                retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
-
             } catch (Exception e) {
                 log.error("[Consumer] Failed to process product-viewed event - offset: {}, key: {}",
                     record.offset(), record.key(), e);
@@ -237,28 +227,20 @@ public void consumeProductViewed(
 
     /**
      * 실패한 레코드 처리 공통 메서드
+     * - 복구 불가능한 에러: DLQ로 전송
+     * - 일시적 에러: 배치 재처리 (Kafka가 자동으로 재시도)
      */
     private void handleFailedRecord(
         ConsumerRecord<String, String> record,
         Exception exception,
         List<ConsumerRecord<String, String>> failedRecords
     ) {
-        // DLQ 전송 여부 결정
         if (dlqPublisher.shouldSendToDlq(exception)) {
-            // 복구 불가능한 에러 → DLQ로 전송
-            int retryCount = retryTracker.getRetryCount(record.topic(), record.partition(), record.offset());
-            dlqPublisher.publishToDlq(record, exception, retryCount);
-            retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
+            // 복구 불가능한 에러 → 즉시 DLQ로 전송
+            dlqPublisher.publishToDlq(record, exception);
         } else {
-            // 일시적 에러 → 재시도 가능 여부 확인
-            if (retryTracker.canRetry(record.topic(), record.partition(), record.offset())) {
-                failedRecords.add(record);
-            } else {
-                // 최대 재시도 횟수 초과 → DLQ로 전송
-                int retryCount = retryTracker.getRetryCount(record.topic(), record.partition(), record.offset());
-                dlqPublisher.publishToDlq(record, exception, retryCount);
-                retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
-            }
+            // 일시적 에러 → 재시도 대상에 추가
+            failedRecords.add(record);
         }
     }
 }

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

@@ -78,10 +78,9 @@ void shouldSaveFailedMessageToDlq() {
             "{\"invalid\":\"json}"
         );
         JsonProcessingException exception = new JsonProcessingException("Parse error") {};
-        int retryCount = 3;
 
         // when
-        dlqPublisher.publishToDlq(record, exception, retryCount);
+        dlqPublisher.publishToDlq(record, exception);
 
         // then
         DlqMessage savedMessage = dlqMessageRepository.findAll().get(0);
@@ -91,7 +90,6 @@ void shouldSaveFailedMessageToDlq() {
         assertThat(savedMessage.getMessageKey()).isEqualTo("product-123");
         assertThat(savedMessage.getMessageValue()).isEqualTo("{\"invalid\":\"json}");
         assertThat(savedMessage.getErrorType()).isEqualTo("JsonProcessingException");
-        assertThat(savedMessage.getRetryCount()).isEqualTo(3);
         assertThat(savedMessage.isResolved()).isFalse();
     }
 
@@ -109,7 +107,7 @@ void shouldIncludeStackTrace() {
         Exception exception = new RuntimeException("Test error");
 
         // when
-        dlqPublisher.publishToDlq(record, exception, 1);
+        dlqPublisher.publishToDlq(record, exception);
 
         // then
         DlqMessage savedMessage = dlqMessageRepository.findAll().get(0);