4장: Kafka 프로듀서와 컨슈머 고급 패턴

프로듀서 → 브로커: 메시지 전송
브로커: 기록 성공 (offset 42)
브로커 → 프로듀서: 응답 전송 (네트워크에서 유실)
프로듀서: 타임아웃 → 재전송
브로커: 동일 메시지 다시 기록 (offset 43) ← 중복!

프로듀서(PID=1) → 브로커: 메시지(seq=0) → 기록
프로듀서(PID=1) → 브로커: 메시지(seq=1) → 기록
프로듀서(PID=1) → 브로커: 메시지(seq=1) → 중복 감지, 거부(DuplicateSequenceException)

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("enable.idempotence", "true"); // 멱등성 활성화
// acks=all, retries=MAX_INT, max.in.flight.requests.per.connection<=5 자동 설정
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

1. 소스 토픽에서 메시지 읽기 (offset 10)
2. 처리
3. 결과를 대상 토픽에 쓰기 ← 성공
4. 오프셋 커밋 (offset 11) ← 실패 (장애 발생)
 
→ 재시작 후 offset 10부터 다시 읽어 중복 처리 발생

Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("transactional.id", "order-processor-1"); // 트랜잭션 ID
props.put("enable.idempotence", "true");
 
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions(); // 트랜잭션 초기화
 
try {
    producer.beginTransaction();
 
    // 여러 토픽/파티션에 쓰기
    producer.send(new ProducerRecord<>("orders-processed", key, value));
    producer.send(new ProducerRecord<>("notifications", key, notification));
 
    // 소스 토픽의 오프셋도 트랜잭션에 포함
    producer.sendOffsetsToTransaction(
        offsets, consumerGroupMetadata
    );
 
    producer.commitTransaction(); // 모두 성공 또는 모두 실패
} catch (Exception e) {
    producer.abortTransaction(); // 롤백
}

// 컨슈머 측: 커밋된 트랜잭션의 메시지만 읽기
Properties consumerProps = new Properties();
consumerProps.put("isolation.level", "read_committed");
// read_committed: 커밋된 트랜잭션 메시지만 소비
// read_uncommitted(기본): 모든 메시지 소비 (중단된 트랜잭션 포함)

Properties props = new Properties();
props.put("enable.auto.commit", "false"); // 자동 커밋 비활성화
 
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("orders"));
 
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
 
    for (ConsumerRecord<String, String> record : records) {
        // 1. 메시지 처리
        processRecord(record);
    }
 
    // 2. 처리 완료 후 수동 커밋
    consumer.commitSync(); // 동기 커밋 (느리지만 안전)
    // consumer.commitAsync(); // 비동기 커밋 (빠르지만 실패 시 재시도 복잡)
}

for (ConsumerRecord<String, String> record : records) {
    processRecord(record);
 
    // 처리된 레코드의 다음 오프셋을 커밋
    Map<TopicPartition, OffsetAndMetadata> offset = Map.of(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1)
    );
    consumer.commitSync(offset);
}

Properties props = new Properties();
 
// 배치 크기 (바이트) — 이 크기가 차면 즉시 전송
props.put("batch.size", 32768); // 32KB (기본: 16KB)
 
// 최대 대기 시간 (밀리초) — 배치가 차지 않아도 이 시간이 지나면 전송
props.put("linger.ms", 20); // 20ms (기본: 0)

// 압축 알고리즘 선택
props.put("compression.type", "lz4");
// none: 압축 안 함 (기본)
// gzip: 높은 압축률, CPU 부하 큼
// snappy: 중간 압축률, 빠른 속도
// lz4: 낮은 압축률, 매우 빠른 속도 (권장)
// zstd: 높은 압축률, lz4보다 약간 느림

private static final int MAX_RETRIES = 3;
private static final String DLQ_TOPIC = "orders.dlq";
 
public void processWithDLQ(ConsumerRecord<String, String> record) {
    int retryCount = 0;
 
    while (retryCount < MAX_RETRIES) {
        try {
            processRecord(record);
            return; // 성공
        } catch (RetriableException e) {
            retryCount++;
            log.warn("처리 실패, 재시도 {}/{}", retryCount, MAX_RETRIES);
            backoff(retryCount); // 지수 백오프
        } catch (NonRetriableException e) {
            break; // 재시도 불필요한 오류
        }
    }
 
    // 재시도 한도 초과 또는 재시도 불가 오류 → DLQ로 전송
    sendToDLQ(record);
}
 
private void sendToDLQ(ConsumerRecord<String, String> record) {
    ProducerRecord<String, String> dlqRecord = new ProducerRecord<>(
        DLQ_TOPIC,
        record.key(),
        record.value()
    );
    // 원본 메타데이터를 헤더에 보존
    dlqRecord.headers().add("original-topic", record.topic().getBytes());
    dlqRecord.headers().add("original-partition",
        String.valueOf(record.partition()).getBytes());
    dlqRecord.headers().add("original-offset",
        String.valueOf(record.offset()).getBytes());
    dlqRecord.headers().add("failure-reason", /* 오류 메시지 */);
 
    dlqProducer.send(dlqRecord);
}