10장: Exactly-once 보장과 신뢰성

1. 메시지 수신 (offset 10)
2. 오프셋 커밋 (offset 11) ← 처리 전에 커밋
3. 메시지 처리 ← 여기서 장애 발생!
4. 재시작 후 offset 11부터 읽기 → offset 10 메시지 유실

1. 메시지 수신 (offset 10)
2. 메시지 처리 ← 성공
3. 오프셋 커밋 (offset 11) ← 여기서 장애 발생!
4. 재시작 후 offset 10부터 읽기 → offset 10 메시지 중복 처리

StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();
 
// 체크포인트 활성화
env.enableCheckpointing(30_000, CheckpointingMode.EXACTLY_ONCE);
 
// Kafka Source (Exactly-once 읽기)
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("orders")
    .setGroupId("flink-processor")
    .setStartingOffsets(OffsetsInitializer.committedOffsets())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();
 
// Kafka Sink (Exactly-once 쓰기)
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(
        KafkaRecordSerializationSchema.builder()
            .setTopic("orders-processed")
            .setValueSerializationSchema(new SimpleStringSchema())
            .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // Exactly-once 보장
    .setTransactionalIdPrefix("flink-orders")              // 트랜잭션 ID 접두사
    .build();
 
DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka");
stream.map(this::processOrder).sinkTo(sink);

// Elasticsearch: 문서 ID 기반 upsert → 자연스러운 멱등 쓰기
// Redis: SET 명령 → 동일 키에 대한 반복 쓰기는 멱등
// DB: INSERT ... ON CONFLICT DO UPDATE → 멱등 upsert

-- PostgreSQL: 주문 ID를 기준으로 upsert
INSERT INTO order_summary (order_id, total_amount, status, updated_at)
VALUES ('O-001', 50000, 'completed', NOW())
ON CONFLICT (order_id)
DO UPDATE SET
    total_amount = EXCLUDED.total_amount,
    status = EXCLUDED.status,
    updated_at = EXCLUDED.updated_at;

def process_event(event):
    event_id = event["eventId"]
 
    # 이미 처리된 이벤트인지 확인
    if redis_client.sismember("processed_events", event_id):
        return  # 중복 → 무시
 
    # 처리 수행
    execute_business_logic(event)
 
    # 처리 완료 기록 (TTL 설정으로 무한 증가 방지)
    redis_client.sadd("processed_events", event_id)
    redis_client.expire("processed_events", 86400)  # 24시간 보존

-- 현재 버전보다 높은 경우에만 업데이트
UPDATE orders
SET status = 'shipped', version = 5
WHERE order_id = 'O-001' AND version = 4;
-- 영향받은 행이 0이면 이미 처리됨 (중복 또는 경합)

프로듀서: 10,000 msg/sec →
Kafka: 충분한 용량 →
처리 엔진: 8,000 msg/sec (병목) →
싱크: 5,000 msg/sec (더 큰 병목)
 
→ 처리 엔진과 싱크에서 백프레셔 발생
→ 처리되지 않은 메시지가 계속 쌓임

Properties props = new Properties();
props.put("max.poll.records", "100");    // poll당 최대 레코드 수 제한
props.put("max.poll.interval.ms", "300000"); // 처리 시간 한도 (5분)
props.put("fetch.max.bytes", "1048576"); // fetch당 최대 바이트 (1MB)

Source → [buffer: 8/10] → Map → [buffer: 10/10 FULL] → Aggregate → Sink
                                      ↑
                            백프레셔 신호 역전파
                     Source의 전송 속도가 자동으로 감소

class DLQProcessor:
    def __init__(self, max_retries=3, retry_delay_base=60):
        self.max_retries = max_retries
        self.retry_delay_base = retry_delay_base
 
    def process_dlq_message(self, message):
        retry_count = int(message.headers.get("retry-count", 0))
        failure_reason = message.headers.get("failure-reason", "")
 
        # 재시도 가능한 오류인지 판단
        if self.is_retriable(failure_reason) and retry_count < self.max_retries:
            delay = self.retry_delay_base * (2 ** retry_count)  # 지수 백오프
            self.schedule_retry(message, delay, retry_count + 1)
        else:
            # 재시도 불가 또는 한도 초과 → 영구 실패 큐로 이동
            self.move_to_permanent_dlq(message)
            self.alert_ops_team(message)
 
    def is_retriable(self, reason):
        retriable_patterns = [
            "timeout", "connection refused",
            "too many requests", "service unavailable"
        ]
        return any(p in reason.lower() for p in retriable_patterns)

StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();
 
// 고정 지연 재시작: 최대 3회, 10초 간격
env.setRestartStrategy(
    RestartStrategies.fixedDelayRestart(3, Time.seconds(10)));
 
// 실패율 기반 재시작: 5분 내 3회 실패까지 허용
env.setRestartStrategy(
    RestartStrategies.failureRateRestart(
        3,                     // 허용 실패 횟수
        Time.minutes(5),       // 측정 기간
        Time.seconds(10)       // 재시작 간 대기
    ));
 
// 지수 백오프 재시작
env.setRestartStrategy(
    RestartStrategies.exponentialDelayRestart(
        Time.seconds(1),       // 초기 대기
        Time.minutes(5),       // 최대 대기
        2.0,                   // 배수
        Time.hours(1),         // 리셋 기간
        0.1                    // 지터
    ));

1. 오프셋 리셋
   - 특정 시점으로 오프셋을 되돌려 해당 시점부터 재처리
   - 멱등 싱크 필수
 
2. Flink Savepoint
   - Savepoint에서 복원하여 해당 시점부터 재처리
   - 작업 수정 후 재배포에 유용
 
3. 새 컨슈머 그룹
   - 새 그룹 ID로 처음부터 읽기
   - 전체 재처리가 필요한 경우