# ReqAdd Kafka Integration Guide

## 📋 개요

Middle Proxy에서 `/auth-user/api/arms/reqAdd` 요청을 Kafka로 발행하여 Backend-Core에서 비동기 처리하는 구조

## 🔄 메시지 플로우

```
Frontend
   │
   │ HTTP POST /auth-user/api/arms/reqAdd/T_ARMS_REQADD_123/addNode.do
   │ Body: { "c_title": "새 요구사항", "ref": 1, ... }
   ▼
Middle-Proxy (Producer)
   │ ReqAddProxyController
   │ kafkaTemplate.send("REQADD", "T_ARMS_REQADD_123", message)
   ▼
Kafka Cluster
   │ Topic: REQADD
   │ Partitions: 3
   │ Key: changeReqTableName (같은 테이블은 같은 파티션)
   ▼
Backend-Core (Consumer)
   │ @KafkaListener(topics = "REQADD")
   │ switch (operation)
   │   case ADD_NODE    → reqAddService.addNode()
   │   case UPDATE_NODE → reqAddService.updateNode()
   │   case REMOVE_NODE → reqAddService.removeNode()
   │   case MOVE_NODE   → reqAddService.moveNode()
   ▼
Database
```

---

## 📦 생성된 파일

### 1. ReqAddKafkaMessage.java
```
경로: src/main/java/com/arms/api/kafka/reqadd/model/ReqAddKafkaMessage.java
역할: Kafka 메시지 포맷 정의
```

**필드:**
- `operation`: ADD_NODE, UPDATE_NODE, REMOVE_NODE, MOVE_NODE
- `changeReqTableName`: 테이블명 (파티션 키로 사용)
- `method`: HTTP 메서드 (참고용)
- `payload`: 요청 본문 (JSON String)
- `timestamp`: 메시지 생성 시간
- `requestId`: 요청 추적 ID (UUID)

### 2. ReqAddProxyController.java
```
경로: src/main/java/com/arms/api/kafka/reqadd/controller/ReqAddProxyController.java
역할: HTTP 요청을 받아 Kafka로 발행
```

**엔드포인트:**
- `POST   /auth-user/api/arms/reqAdd/{changeReqTableName}/addNode.do`
- `POST   /auth-user/api/arms/reqAdd/{changeReqTableName}/updateNode.do`
- `DELETE /auth-user/api/arms/reqAdd/{changeReqTableName}/removeNode.do`
- `POST   /auth-user/api/arms/reqAdd/{changeReqTableName}/moveNode.do`

### 3. AppConfig.java (수정)
```
경로: src/main/java/com/arms/config/AppConfig.java
추가: ObjectMapper Bean
```

---

## 🧪 테스트 방법

### 1. 노드 추가 (addNode)

```bash
curl -X POST "http://localhost:13131/auth-user/api/arms/reqAdd/T_ARMS_REQADD_123/addNode.do" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "ref": 1,
    "c_position": 0,
    "c_title": "새로운 요구사항",
    "c_type": "default"
  }'
```

**예상 응답 (202 ACCEPTED):**
```json
{
  "status": "ACCEPTED",
  "message": "Request queued for processing",
  "requestId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "operation": "ADD_NODE",
  "table": "T_ARMS_REQADD_123",
  "timestamp": 1700000000000
}
```

### 2. 노드 수정 (updateNode)

```bash
curl -X POST "http://localhost:13131/auth-user/api/arms/reqAdd/T_ARMS_REQADD_123/updateNode.do" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "c_id": 123,
    "c_title": "수정된 제목"
  }'
```

### 3. 노드 삭제 (removeNode)

```bash
curl -X DELETE "http://localhost:13131/auth-user/api/arms/reqAdd/T_ARMS_REQADD_123/removeNode.do" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "c_id": 123
  }'
```

### 4. 노드 이동 (moveNode)

```bash
curl -X POST "http://localhost:13131/auth-user/api/arms/reqAdd/T_ARMS_REQADD_123/moveNode.do" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "c_id": 123,
    "ref": 2,
    "c_position": 1,
    "copy": 0
  }'
```

---

## 🔧 Backend-Core Consumer 구현 예시

Backend-Core에서 다음과 같이 Consumer를 구현해야 합니다:

```java
@Service
@Slf4j
public class ReqAddKafkaConsumer {

    @Autowired
    private ReqAddService reqAddService;
    
    @Autowired
    private ObjectMapper objectMapper;

    @KafkaListener(
        topics = "REQADD",
        groupId = "backend-core-reqadd-consumer",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void consume(String message) {
        
        try {
            // JSON 역직렬화
            ReqAddKafkaMessage kafkaMessage = 
                objectMapper.readValue(message, ReqAddKafkaMessage.class);
            
            log.info("📥 Received Kafka Message | Operation: {}, Table: {}, RequestId: {}",
                kafkaMessage.getOperation(),
                kafkaMessage.getChangeReqTableName(),
                kafkaMessage.getRequestId());
            
            // Payload 역직렬화
            String payload = kafkaMessage.getPayload();
            
            // Operation에 따라 분기
            switch (kafkaMessage.getOperation()) {
                case "ADD_NODE":
                    reqAddService.addNode(
                        kafkaMessage.getChangeReqTableName(), 
                        payload
                    );
                    break;
                    
                case "UPDATE_NODE":
                    reqAddService.updateNode(
                        kafkaMessage.getChangeReqTableName(), 
                        payload
                    );
                    break;
                    
                case "REMOVE_NODE":
                    reqAddService.removeNode(
                        kafkaMessage.getChangeReqTableName(), 
                        payload
                    );
                    break;
                    
                case "MOVE_NODE":
                    reqAddService.moveNode(
                        kafkaMessage.getChangeReqTableName(), 
                        payload
                    );
                    break;
                    
                default:
                    log.warn("⚠️ Unknown operation: {}", kafkaMessage.getOperation());
            }
            
            log.info("✅ Message processed successfully | RequestId: {}", 
                kafkaMessage.getRequestId());
            
        } catch (Exception e) {
            log.error("❌ Failed to process Kafka message", e);
            // DLQ로 전송하거나 재처리 로직 추가
        }
    }
}
```

---

## 📊 Kafka 모니터링

### 1. Topic 확인
```bash
kafka-topics.sh --bootstrap-server localhost:9094 \
  --describe --topic REQADD
```

### 2. Consumer Lag 확인
```bash
kafka-consumer-groups.sh --bootstrap-server localhost:9094 \
  --describe --group backend-core-reqadd-consumer
```

### 3. 메시지 조회 (개발 환경)
```bash
kafka-console-consumer.sh --bootstrap-server localhost:9094 \
  --topic REQADD \
  --from-beginning \
  --property print.key=true \
  --property print.timestamp=true
```

---

## 🚨 에러 처리

### 1. Kafka 발행 실패
- **현상**: 500 Internal Server Error 응답
- **원인**: Kafka 클러스터 연결 실패
- **해결**: Kafka 서버 상태 확인, `bootstrap-servers` 설정 확인

### 2. JSON 직렬화 실패
- **현상**: 400 Bad Request 응답
- **원인**: 잘못된 JSON 포맷
- **해결**: 요청 본문 형식 확인

### 3. Consumer 처리 실패
- **현상**: 메시지가 Kafka에는 있으나 DB 반영 안됨
- **원인**: Backend-Core Consumer 오류
- **해결**: Backend-Core 로그 확인

---

## 🎯 주의사항

### 1. 순서 보장
- 같은 `changeReqTableName`에 대한 메시지는 같은 파티션으로 전송됨
- 파티션 내에서는 순서가 보장됨

### 2. 멱등성 (Idempotence)
- Producer에서 `enable.idempotence=true` 설정됨
- 중복 메시지 방지

### 3. 재시도 (Retry)
- Producer 재시도: 3회
- Consumer 재시도: DLQ(Dead Letter Queue) 구현 권장

### 4. 응답 시간
- 클라이언트는 즉시 202 ACCEPTED 응답 받음
- 실제 처리는 Backend-Core에서 비동기로 수행
- 처리 결과 확인이 필요한 경우 별도 조회 API 또는 WebSocket 사용

---

## 📝 로그 예시

### Middle-Proxy (Producer)
```
2024-11-25 10:30:45.123 INFO  [reqadd-proxy] 📤 Kafka Message Published | Topic: REQADD, Operation: ADD_NODE, Table: T_ARMS_REQADD_123, Partition: 1, Offset: 12345, RequestId: a1b2c3d4-e5f6-7890-abcd-ef1234567890
```

### Backend-Core (Consumer)
```
2024-11-25 10:30:45.234 INFO  [reqadd-consumer] 📥 Received Kafka Message | Operation: ADD_NODE, Table: T_ARMS_REQADD_123, RequestId: a1b2c3d4-e5f6-7890-abcd-ef1234567890
2024-11-25 10:30:45.567 INFO  [reqadd-consumer] ✅ Message processed successfully | RequestId: a1b2c3d4-e5f6-7890-abcd-ef1234567890
```

---

## ✅ 체크리스트

### Middle-Proxy (완료)
- [x] ReqAddKafkaMessage DTO 생성
- [x] ReqAddProxyController 생성
- [x] ObjectMapper Bean 추가
- [x] KafkaConfig 확인

### Backend-Core (작업 필요)
- [ ] ReqAddKafkaConsumer 생성
- [ ] KafkaConfig Consumer 설정 추가
- [ ] 기존 ReqAddService와 통합
- [ ] DLQ(Dead Letter Queue) 구현
- [ ] 에러 처리 및 재시도 로직 추가

---

## 🔗 관련 문서
- [Kafka 공식 문서](https://kafka.apache.org/documentation/)
- [Spring Kafka 문서](https://spring.io/projects/spring-kafka)
- [A-RMS 아키텍처 문서](./README.md)