[FEAT] 회원 탈퇴로 인한 서비스 전파 fix : 토픽 이름 변경, dlt 로깅, 일부 로직 변경, yml 파일 수정

Blueline0425 · Blueline0425 · commit 6a950201de7c · 2025-10-18T01:29:27.000+09:00
diff --git a/src/main/java/com/project/user/domain/application/usecase/DeleteAccountUseCase.java b/src/main/java/com/project/user/domain/application/usecase/DeleteAccountUseCase.java
@@ -6,6 +6,7 @@
 import com.project.user.domain.domain.service.UserService;
 import com.project.user.global.exception.RestApiException;
 import com.project.user.global.security.TokenProvider;
+import com.project.user.domain.infra.kafka.ResilientSender;
 import jakarta.transaction.Transactional;
 import lombok.RequiredArgsConstructor;
 import org.mindrot.jbcrypt.BCrypt;
@@ -27,9 +28,8 @@ public class DeleteAccountUseCase {
 
     private final TokenProvider tokenProvider;
     private final UserService userService;
-    private final KafkaTemplate<String, String> kafka;  // ★ 카프카 프로듀서
     private final ObjectMapper om;
-
+    private final ResilientSender sender;
     /**
      * 유저 삭제 흐름의 시작점.
      * 1) 액세스 토큰에서 userNo 추출 및 사용자 검증
@@ -63,10 +63,11 @@ public void execute(String accessToken, UserDeletionRequest request) {
         try {
             String payload = om.writeValueAsString(start);
             // key=userNo 로 파티셔닝 → 동일 유저 이벤트 순서 보장
-            kafka.send("user.deletion.start", String.valueOf(userNo), payload).get();
+            //kafka.send("user.start-delete.command", String.valueOf(userNo), payload).get();
+            sender.sendSync("user.start-delete.command", String.valueOf(userNo), payload);
         } catch (Exception e) {
             // 프로듀스 실패 → 트랜잭션 롤백(서비스 계층 예외로 래핑)
-            throw new RuntimeException("failed to publish user.deletion.start", e);
+            throw new RuntimeException("failed to publish user.start-delete.command", e);
         }
     }
 
diff --git a/src/main/java/com/project/user/domain/infra/kafka/DltLogger.java b/src/main/java/com/project/user/domain/infra/kafka/DltLogger.java
@@ -0,0 +1,43 @@
+package com.project.user.domain.infra.kafka;
+// 공통 유틸(선택): 헤더를 예쁘게 로깅하고, 카운터 증가
+
+import io.micrometer.core.instrument.MeterRegistry;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class DltLogger {
+
+    private final MeterRegistry meter; // micrometer
+
+    public void logAndCount(
+            String dltTopic,
+            ConsumerRecord<String, String> rec,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_TOPIC, required = false)   String oTopic,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_PARTITION, required = false) Integer oPart,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_OFFSET, required = false)   Long oOffset,
+            @Header(name = KafkaHeaders.DLT_EXCEPTION_FQCN, required = false)    String exClass,
+            @Header(name = KafkaHeaders.DLT_EXCEPTION_MESSAGE, required = false) String exMsg
+    ) {
+        // 메트릭: dlt 카운트
+        meter.counter("kafka.dlt.count",
+                        "dltTopic", dltTopic,
+                        "origTopic", String.valueOf(oTopic))
+                .increment();
+
+        log.error("[DLT] topic={}, origTopic={}, origPartition={}, origOffset={}, key={}, ts={}, exClass={}, exMsg={}, payload={}",
+                dltTopic,
+                oTopic, oPart, oOffset,
+                rec.key(),
+                rec.timestamp(),
+                exClass, exMsg,
+                rec.value()
+        );
+    }
+}
diff --git a/src/main/java/com/project/user/domain/infra/kafka/ResilientSender.java b/src/main/java/com/project/user/domain/infra/kafka/ResilientSender.java
@@ -0,0 +1,28 @@
+package com.project.user.domain.infra.kafka;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+import io.github.resilience4j.retry.annotation.Retry;
+import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
+
+/**
+ * 카프카 전송 래퍼:
+ * - .get() 호출로 브로커 acks=all 까지 동기 확인(실패 시 예외 발생)
+ * - Resilience4j @Retry/@CircuitBreaker 로 일시 장애 완충
+ *   (Fallback 은 재던져서 Outbox 백오프로 연결)
+ */
+@Component @RequiredArgsConstructor
+public class ResilientSender {
+    private final KafkaTemplate<String,String> kafka;
+
+    @Retry(name="kafkaSend")
+    @CircuitBreaker(name="kafkaSend", fallbackMethod = "sendFallback")
+    public void sendSync(String topic, String key, String payload) throws Exception {
+        kafka.send(topic, key, payload).get(); // 동기 전송
+    }
+    @SuppressWarnings("unused")
+    public void sendFallback(String topic, String key, String payload, Throwable t) throws Exception {
+        throw (t instanceof Exception) ? (Exception)t : new RuntimeException(t);
+    }
+}
diff --git a/src/main/java/com/project/user/domain/infra/kafka/UserDltListeners.java b/src/main/java/com/project/user/domain/infra/kafka/UserDltListeners.java
@@ -0,0 +1,27 @@
+package com.project.user.domain.infra.kafka;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class UserDltListeners {
+
+    private final DltLogger dlt;
+
+    @KafkaListener(topics = "user.final-delete.command.dlt", groupId = "user-service")
+    public void onFinalDeleteCmdDlt(
+            ConsumerRecord<String,String> rec,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_TOPIC, required = false)   String oTopic,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_PARTITION, required = false) Integer oPart,
+            @Header(name = KafkaHeaders.DLT_ORIGINAL_OFFSET, required = false)   Long oOffset,
+            @Header(name = KafkaHeaders.DLT_EXCEPTION_FQCN, required = false)    String exClass,
+            @Header(name = KafkaHeaders.DLT_EXCEPTION_MESSAGE, required = false) String exMsg
+    ){
+        dlt.logAndCount("user.final-delete.command.dlt", rec, oTopic, oPart, oOffset, exClass, exMsg);
+    }
+}
diff --git a/src/main/java/com/project/user/domain/infra/kafka/UserFinalDeleteListener.java b/src/main/java/com/project/user/domain/infra/kafka/UserFinalDeleteListener.java
@@ -10,7 +10,6 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.annotation.RetryableTopic;
-import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.stereotype.Component;
@@ -23,9 +22,8 @@ public class UserFinalDeleteListener {
 
     private final UserRepository userRepo;
     private final ProcessedEventRepository peRepo;
-    private final KafkaTemplate<String,String> kafka;
     private final ObjectMapper om;
-
+    private final ResilientSender sender;
     /** 멱등 시작 마킹: (없으면 INSERT) → 이미 SUCCESS면 스킵 */
     private boolean tryBegin(String eventId, String type){
         if (!peRepo.existsById(eventId)) {
@@ -61,81 +59,84 @@ private void markError(String eventId, String msg){
 
     /**
      * 최종 삭제 처리 플로우:
-     * - @RetryableTopic: 최대 5회 지수 백오프 재시도 → 실패 시 user.final.delete.dlt 로 이동
+     * - @RetryableTopic: 최대 5회 지수 백오프 재시도 → 실패 시 user.final-delete.command.dlt 로 이동
      * - InvalidPayloadException 은 exclude → 즉시 DLT(재시도 없음)
      * - 수신 → 파싱/검증(독성은 즉시 DLT) → 멱등 시작 → 비즈니스(하드 삭제) → reply 동기 전송
      * - 커밋 이후에만 수동 ACK(MANUAL_IMMEDIATE)로 오프셋 커밋 → DB/메시지 정합 보장
      */
     @RetryableTopic(
             attempts = "5",
             backoff = @Backoff(delay = 1000, multiplier = 2.0),
-            autoCreateTopics = "true",
+            autoCreateTopics = "false",
             dltTopicSuffix = ".dlt",
-            exclude = { InvalidPayloadException.class } // ★ 독성 페이로드는 즉시 DLT
+            exclude = { InvalidPayloadException.class }
     )
     @KafkaListener(
-            topics = "user.final.delete",
+            topics = "user.final-delete.command",
             groupId = "user-service",
             containerFactory = "kafkaManualAckFactory"
     )
     @Transactional
     public void onFinalDelete(ConsumerRecord<String,String> rec, Acknowledgment ack) throws Exception {
 
-        // 0) 파싱 + 유효성 검증 — 독성(InvalidPayloadException)은 즉시 DLT 보내도록 로깅 후 재던짐
+        // 0) 파싱 + 유효성 검증 — 독성(InvalidPayloadException)은 즉시 DLT
         final JsonNode n;
         try {
             n = om.readTree(rec.value());
-            String eventId = n.path("eventId").asText("").trim();
-            long userNo    = n.path("userNo").asLong(0L);
-            if (eventId.isEmpty()) throw new InvalidPayloadException("Missing or empty 'eventId'");
-            if (userNo <= 0L)      throw new InvalidPayloadException("Invalid 'userNo'");
+        } catch (Exception e) {
+            log.error("[user-service] toxic payload -> DLT, value={}", rec.value());
+            throw new InvalidPayloadException("Invalid JSON: " + e.getMessage(), e);
+        }
 
-            // 1) 멱등 처리(이미 성공이면 스킵)
-            if (!tryBegin(eventId, "FINAL_DELETE")) {
-                ack.acknowledge();
-                return;
-            }
+        final String eventId = n.path("eventId").asText("").trim();
+        final long userNo    = n.path("userNo").asLong(0L);
+        if (eventId.isEmpty()) throw new InvalidPayloadException("Missing or empty 'eventId'");
+        if (userNo <= 0L)      throw new InvalidPayloadException("Invalid 'userNo'");
 
-            try {
-                // 2) 실제 하드 삭제(멱등)
-                userRepo.deleteById(userNo);
+        // 1) 멱등 처리(이미 SUCCESS면 스킵)
+        if (!tryBegin(eventId, "FINAL_DELETE")) {
+            ack.acknowledge();
+            return;
+        }
 
-                // 3) 성공 reply 동기 전송(acks=all, .get())
-                var reply = om.createObjectNode()
-                        .put("eventId", eventId)
-                        .put("userNo", userNo)
-                        .put("status", "SUCCESS")
-                        .put("type", "FINAL_DELETE");
-                kafka.send("user.final.reply", String.valueOf(userNo), om.writeValueAsString(reply)).get();
+        // 2) 실제 하드 삭제(멱등)
+        try {
+            userRepo.deleteById(userNo);
 
-                // 운영 관측성: 키/이벤트/유저
-                log.info("FINAL_DELETE ok key={}, eventId={}, userNo={}", rec.key(), eventId, userNo);
+            // 3) 성공 마킹(DB 커밋 대상)
+            markSuccess(eventId);
 
-                // 4) 성공 마킹
-                markSuccess(eventId);
+            // 4) 커밋 이후 reply 전송 + ACK
+            final String replyPayload = om.createObjectNode()
+                    .put("eventId", eventId)
+                    .put("userNo", userNo)
+                    .put("status", "SUCCESS")
+                    .put("type", "FINAL_DELETE")
+                    .toString();
 
-                // 5) 트랜잭션 커밋 이후 ACK (정합성 보장)
-                org.springframework.transaction.support.TransactionSynchronizationManager.registerSynchronization(
-                        new org.springframework.transaction.support.TransactionSynchronization() {
-                            @Override public void afterCommit() { ack.acknowledge(); }
+            org.springframework.transaction.support.TransactionSynchronizationManager.registerSynchronization(
+                    new org.springframework.transaction.support.TransactionSynchronization() {
+                        @Override public void afterCommit() {
+                            try {
+                                // 커밋 확정 후에만 브로커에 회신(정합성 ↑)
+                                sender.sendSync("user.final-delete.reply", String.valueOf(userNo), replyPayload);
+                                ack.acknowledge(); // 전송 성공시에만 오프셋 커밋
+                                log.info("FINAL_DELETE ok key={}, eventId={}, userNo={}", rec.key(), eventId, userNo);
+                            } catch (Exception sendEx) {
+                                // 전송 실패: ACK 하지 않음 → 오프셋 미커밋 상태 유지
+                                // 컨슈머 재시작/리밸런스 시 같은 레코드 재처리됨(@RetryableTopic와는 별개)
+                                log.error("FINAL_DELETE reply send failed (will be retried on re-consume). " +
+                                        "eventId={}, userNo={}, err={}", eventId, userNo, sendEx.toString());
+                            }
                         }
-                );
-            } catch (Exception ex) {
-                log.warn("FINAL_DELETE failed userNo={}, eventId={}, err={}", userNo, eventId, ex.toString());
-                markError(eventId, ex.getMessage());
-                throw ex; // 재시도 → 최대 횟수 후 DLT
-            }
+                    }
+            );
 
-        } catch (InvalidPayloadException bad) {
-            // ★ 독성 페이로드 → 재시도 없이 즉시 DLT (exclude 규칙)
-            log.error("[user-service] toxic payload -> DLT, value={}", rec.value());
-            throw bad;
+        } catch (Exception ex) {
+            // 비즈니스/기술 예외 → 재시도/DTL 체인
+            log.warn("FINAL_DELETE failed userNo={}, eventId={}, err={}", userNo, eventId, ex.toString());
+            markError(eventId, ex.getMessage());
+            throw ex;
         }
     }
-
-    @KafkaListener(topics = "user.final.delete.dlt", groupId = "user-service")
-    public void onFinalDeleteDlt(String payload){
-        // DLT 모니터링(알람/대시보드 연계 지점)
-        log.error("[DLT][user.final.delete] {}", payload);
-    }
-}
+}
diff --git a/src/main/java/com/project/user/global/config/KafkaConfig.java b/src/main/java/com/project/user/global/config/KafkaConfig.java
@@ -24,9 +24,9 @@ public ProducerFactory<String,String> producerFactory(KafkaProperties props){
         Map<String,Object> cfg = new HashMap<>(props.buildProducerProperties());
         cfg.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         cfg.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        cfg.put(ProducerConfig.ACKS_CONFIG, "all");                 // 브로커 전부 ACK 요구
-        cfg.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);    // 멱등 프로듀서
-        cfg.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); // 순서 보장
+//        cfg.put(ProducerConfig.ACKS_CONFIG, "all");                 // 브로커 전부 ACK 요구
+//        cfg.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);    // 멱등 프로듀서
+//        cfg.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); // 순서 보장
         return new DefaultKafkaProducerFactory<>(cfg);
     }
     @Bean public KafkaTemplate<String,String> kafkaTemplate(ProducerFactory<String,String> pf){ return new KafkaTemplate<>(pf); }
@@ -37,7 +37,7 @@ public ConsumerFactory<String,String> consumerFactory(KafkaProperties props){
         cfg.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
         cfg.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
         cfg.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);   // 수동 커밋(ACK)
-        cfg.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // 운영 기본: 최신부터
+        cfg.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // 클라우드 운영 기본: 최신부터
         cfg.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);       // 배치 사이즈(사건당 작업시간 고려)
         cfg.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300_000); // 최대 처리 간격(5분)
         return new DefaultKafkaConsumerFactory<>(cfg);
@@ -58,7 +58,7 @@ public ConcurrentKafkaListenerContainerFactory<String,String> kafkaManualAckFact
         var f = new ConcurrentKafkaListenerContainerFactory<String,String>();
         f.setConsumerFactory(cf);
         f.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); // 우리가 직접 ack
-        f.setConcurrency(1); // 파티션 1개 기준(필요 시 조정)
+        //f.setConcurrency(1); // 파티션 1개 기준(필요 시 조정)
         return f;
     }
 
diff --git a/src/main/java/com/project/user/global/config/Resilience4jConfig.java b/src/main/java/com/project/user/global/config/Resilience4jConfig.java
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
@@ -24,11 +24,28 @@ resilience4j:
 
 spring:
   kafka:
-    bootstrap-servers: ${SPRING_KAFKA_BOOTSTRAP_SERVERS:localhost:9094} # 기본값은 로컬 외부포트
+    bootstrap-servers: pkc-gq2xn.asia-northeast3.gcp.confluent.cloud:9092
+    properties:
+      security.protocol: SASL_SSL
+      sasl.mechanism: PLAIN
+      sasl.jaas.config: >
+        org.apache.kafka.common.security.plain.PlainLoginModule required
+        username="AGQDVMNB6DVPXFX7"
+        password="cfltBaumbQnzwuPIRS3E8S1MNrfqrsb9LXzDvUBPBrP0YQRUTfBqdD2nOKmzUXFw";
+    producer:
+      acks: all
+      retries: 3
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
     consumer:
-      group-id: user-service
+      enable-auto-commit: false
+      auto-offset-reset: latest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
     admin:
+      auto-create: false
       fail-fast: true
+      
   datasource:
     username: ${RDB_USERNAME}
     url: ${RDB_URL}
