Backend Deep Dive #1: Kafka는 왜 유실되지 않았을까? CAP 이론으로 본 메시지 시스템 구조

// 이 코드는 개념 이해를 돕기 위한 의사 코드(Pseudo-code)에 가까우며,
// 실제 실행을 위해서는 RabbitMQ, Kafka 클라이언트 라이브러리 의존성을 추가하고
// 각 메시지 큐 서버가 실행 중이어야 한다. (예: build.gradle.kts)
// implementation("com.rabbitmq:amqp-client:5.14.2")
// implementation("org.apache.kafka:kafka-clients:3.1.0")
 
import com.rabbitmq.client.AMQP
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.DefaultConsumer
import com.rabbitmq.client.Envelope
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.consumer.KafkaConsumer
import java.time.Duration
import java.util.*
 
// --- RabbitMQ 소비자 예제 ---
// '스마트 브로커'와 함께 동작하는 소비자
// 개발자가 직접 Ack/Nack를 제어해야 하는 책임을 가진다.
 
object RabbitMQConsumer {
    private const val QUEUE_NAME = "my_queue"
 
    fun run() {
        val factory = ConnectionFactory()
        factory.host = "localhost"
        val connection = factory.newConnection()
        val channel = connection.createChannel()
 
        channel.queueDeclare(QUEUE_NAME, false, false, false, null)
        println(" [*] 메시지를 기다리는 중. 종료하려면 프로그램을 중단하세요.")
 
        // autoAck=false로 설정하여 수동으로 Ack를 보내도록 설정
        // 이것이 '책임'의 시작점.
        val autoAck = false
 
        val consumer = object : DefaultConsumer(channel) {
            override fun handleDelivery(
                consumerTag: String?,
                envelope: Envelope,
                properties: AMQP.BasicProperties?,
                body: ByteArray
            ) {
                val message = String(body, Charsets.UTF_8)
                println(" [x] 메시지 수신: '$message'")
 
                try {
                    // 일부러 짝수 메시지는 실패하는 상황을 시뮬레이션
                    if (message.toInt() % 2 == 0) {
                        throw RuntimeException("처리 실패: 짝수 메시지")
                    }
                    // 실제 비즈니스 로직 수행
                    Thread.sleep(1000)
                    println(" [v] 메시지 처리 성공: '$message'")
 
                    // 1. 처리 성공 시: 브로커에게 '처리 완료' 확인증(Ack)을 보낸다.
                    // 이 신호를 받은 브로커는 큐에서 메시지를 완전히 삭제한다.
                    channel.basicAck(envelope.deliveryTag, false)
 
                } catch (e: Exception) {
                    println(" [!] ${e.message}")
                    // 2. 처리 실패 시: 브로커에게 '처리 실패' 신호(Nack)를 보낸다.
                    // requeue=true로 설정하면 메시지가 다시 큐로 돌아가 다른 소비자에게 전달된다.
                    channel.basicNack(envelope.deliveryTag, false, true)
                }
            }
        }
        channel.basicConsume(QUEUE_NAME, autoAck, consumer)
    }
}
 
// --- Kafka 소비자 예제 ---
// '멍청한 브로커'와 함께 동작하는 '스마트 컨슈머'
// 소비자가 스스로 어디까지 읽었는지(offset)를 관리한다.
 
object KafkaConsumer {
    private const val TOPIC_NAME = "my_topic"
    private const val GROUP_ID = "my_group"
 
    fun run() {
        val props = Properties()
        props[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = "localhost:9092"
        props[ConsumerConfig.GROUP_ID_CONFIG] = GROUP_ID
        props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = "org.apache.kafka.common.serialization.StringDeserializer"
        props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = "org.apache.kafka.common.serialization.StringDeserializer"
        // enable.auto.commit=true가 기본값이며, 주기적으로 오프셋을 자동으로 커밋합니다.
        // 더 정교한 제어를 위해 false로 두고 수동으로 커밋할 수도 있습니다.
        props[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = "true"
        props[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
 
        val consumer = KafkaConsumer<String, String>(props)
        consumer.subscribe(listOf(TOPIC_NAME))
 
        println("[*] Kafka에서 메시지를 기다리는 중...")
 
        try {
            while (true) {
                // 소비자는 그저 루프를 돌며 메시지를 가져와(poll) 처리하기만 하면 된다.
                val records = consumer.poll(Duration.ofMillis(100))
                for (record in records) {
                    println(
                        "토픽: ${record.topic()}, 파티션: ${record.partition()}, " +
                        "오프셋: ${record.offset()}, 키: ${record.key()}, 값: ${record.value()}"
                    )
                    // 실제 비즈니스 로직 처리
                    // 만약 여기서 에러가 발생하여 소비자가 중단되더라도,
                    // 오프셋이 커밋되지 않았기 때문에 재시작 시 이 메시지부터 다시 처리한다.
                    // 데이터 유실이 발생하지 않는다.
                }
            }
        } catch (e: Exception) {
            println("소비자 에러: ${e.message}")
        } finally {
            consumer.close()
            println("소비자 종료")
        }
    }
}
 
fun main() {
    // 실행하려는 예제의 주석을 해제.
    // RabbitMQConsumer.run()
    // KafkaConsumer.run()
}