Consumer

Kafka 컨슈머란?

Kafka 컨슈머는 Kafka 클러스터에서 토픽의 메시지를 읽고 처리하는 컴포넌트입니다. 컨슈머는 프로듀서가 전송한 메시지를 토픽의 파티션에서 읽어와 다양한 애플리케이션 로직에 맞춰 데이터를 처리합니다.

Kafka는 고성능 비동기 스트리밍을 지원하며, 컨슈머는 데이터를 빠르게 소비할 수 있는 주요 구성 요소입니다.

개념

컨슈머는 Kafka의 토픽으로부터 데이터를 읽어오며, 여러 컨슈머가 컨슈머 그룹을 형성해 병렬 처리를 수행합니다.

각 컨슈머는 파티션을 할당받아 데이터를 처리하며, 컨슈머 그룹 내에서 데이터 중복 없이 각 파티션에 접근합니다. 오프셋은 각 파티션 내에서 컨슈머가 어디까지 데이터를 읽었는지 추적하는 중요한 개념입니다.

이를 통해 중단되거나 장애가 발생하더라도 이어서 데이터를 처리할 수 있습니다.

동작 원리

컨슈머가 Kafka 클러스터에서 데이터를 소비하는 과정은 다음과 같습니다.

• 컨슈머는 토픽을 구독하여 해당 토픽의 메시지를 읽습니다.

• Kafka는 컨슈머 그룹 내에서 파티션을 할당하며, 각 컨슈머가 독립적으로 파티션을 처리하도록 합니다.

• 컨슈머가 파티션에서 메시지를 읽을 때, Kafka는 그 위치를 오프셋으로 추적합니다.

• 메시지가 처리되면 컨슈머는 Kafka에 오프셋을 커밋하여 어디까지 읽었는지를 기록하고, 다음 메시지부터 이어서 처리할 수 있습니다.

상세 설명

Kafka 컨슈머는 리밸런싱이라는 개념을 사용해 그룹 내의 컨슈머가 추가되거나 제거될 때 자동으로 파티션을 재분배합니다. 이렇게 하면 컨슈머가 동적으로 클러스터에 추가되거나 제거되어도 데이터 처리에 영향이 없게 됩니다.

컨슈머는 오프셋 자동 커밋과 수동 커밋 기능을 제공하여, 메시지가 성공적으로 처리되었는지 확실히 한 후에 오프셋을 커밋하는 방식으로 데이터 처리의 일관성을 유지할 수 있습니다.

옵션

Kafka 컨슈머는 다양한 설정 옵션을 제공하여 성능과 안정성을 조정할 수 있습니다.

• enable.auto.commit: 자동 커밋 여부를 설정합니다. 기본값은 true로, 일정 간격마다 Kafka가 오프셋을 자동으로 커밋합니다.

• auto.offset.reset: 컨슈머가 읽기 시작할 오프셋을 설정합니다. 예를 들어, 토픽에 처음 연결될 때 또는 마지막 커밋된 오프셋이 없을 때, earliest를 설정하면 가장 오래된 메시지부터 읽고, latest를 설정하면 최신 메시지부터 읽습니다.

• max.poll.records: 한 번의 요청으로 가져올 수 있는 최대 레코드 수를 정의합니다. 이 값을 조정하면 처리 속도와 메모리 사용량을 최적화할 수 있습니다.

• session.timeout.ms: 컨슈머가 정상적으로 작동 중인지 Kafka에 신호를 보내는 시간 간격입니다. 지정된 시간 동안 컨슈머가 응답하지 않으면 Kafka는 해당 컨슈머가 장애가 발생한 것으로 간주하고 리밸런싱을 시작합니다.

Kafka 컨슈머의 할당 방식

Kafka 컨슈머는 파티션 할당을 통해 메시지를 소비합니다. Kafka는 컨슈머가 어떤 파티션에서 데이터를 읽을지를 결정하는 두 가지 방법을 제공합니다

subscribe(): Kafka가 자동으로 파티션을 할당하는 방식입니다. 컨슈머 그룹 내에서 파티션이 자동으로 분배되며, Kafka가 이를 관리합니다.

assign(): 사용자가 직접 특정 파티션을 컨슈머에 할당하는 방식입니다. 이를 통해 더 세밀하게 파티션을 제어할 수 있습니다.

---

subscribe() 란?

subscribe()는 Kafka 컨슈머가 하나 이상의 토픽을 구독하고, Kafka 클러스터 내에서 자동으로 파티션을 할당받아 메시지를 소비하는 메서드입니다. subscribe()를 사용하면 Kafka가 컨슈머 그룹을 통해 자동으로 파티션을 분배하고, 파티션의 리밸런싱(재할당)도 자동으로 관리합니다.

subscribe()는 일반적으로 대규모 데이터 스트리밍에서 많이 사용됩니다. 특히 여러 컨슈머가 컨슈머 그룹을 형성하여 병렬로 데이터 처리를 할 때 유용합니다.

• 다수의 컨슈머가 하나의 토픽을 동시에 처리할 때, Kafka가 파티션을 자동으로 분배하도록 하고 싶을 때

• 자동 리밸런싱을 통해 컨슈머 그룹 내에서 컨슈머 추가/제거 시 파티션을 자동으로 조정하고 싶을 때

• 동적으로 토픽을 관리하고, 특정 컨슈머가 여러 토픽을 구독하게 하고 싶을 때

assign() 란?

assign() 메서드는 Kafka 컨슈머가 특정 파티션을 직접 할당받아 메시지를 읽을 수 있도록 합니다. 이 메서드를 사용하면 컨슈머 그룹의 자동 리밸런싱을 피하고, 특정 파티션에서 데이터를 수동으로 가져올 수 있습니다.

assign()은 다음과 같은 상황에서 유용합니다

• 특정 파티션만 소비하고 싶을 때: 특정 사용자의 데이터나 특정 지역에 해당하는 데이터를 처리하고 싶을 때, 직접 파티션을 할당할 수 있습니다.

• 리밸런싱을 피하고 싶을 때: 컨슈머 그룹 내에서 리밸런싱이 자주 발생하면 성능 저하가 있을 수 있습니다. assign()을 사용하면 Kafka의 자동 리밸런싱 없이 파티션을 고정할 수 있습니다.

• 컨슈머 그룹 기능을 사용하지 않을 때: 컨슈머 그룹 없이, 단일 컨슈머가 여러 파티션을 처리해야 할 경우에도 assign()을 사용하여 수동으로 파티션을 관리할 수 있습니다.

assign() 메서드를 사용하면 특정 파티션을 직접 지정하여 데이터를 읽을 수 있습니다.

val consumer = KafkaConsumer<String, String>(ClientConfiguration().config())

// 특정 토픽의 파티션을 직접 할당
val partition = TopicPartition("my-topic", 0)
consumer.assign(listOf(partition))

while (true) {
    val records = consumer.poll(Duration.ofMillis(100))
    records.forEach { record ->
        println("Consumed message: ${record.value()} from partition ${record.partition()}")
    }
}

위 예시에서 TopicPartition을 통해 토픽 이름과 파티션 번호를 지정하고, 그 파티션만 소비하도록 assign()을 사용하고 있습니다. 이를 통해 리밸런싱 없이 특정 파티션에서 데이터를 소비할 수 있습니다.

---

assign()과 subscribe()의 차이점

subscribe()와 assign()은 각각 파티션 할당의 자동화와 수동화를 담당합니다. 다음은 두 방식의 주요 차이점입니다

subscribe(): 컨슈머는 Kafka 클러스터 내에서 컨슈머 그룹의 일부로서 동작하며, Kafka는 자동으로 파티션을 할당합니다. 이 방식은 리밸런싱이 발생할 수 있으며, 여러 컨슈머가 동시에 하나의 토픽을 소비할 수 있습니다.

• 장점: 여러 컨슈머가 파티션을 나누어 처리할 수 있으며, Kafka가 파티션을 자동으로 분배하기 때문에 관리가 쉽습니다.

• 단점: 리밸런싱이 빈번하게 발생하면 성능 저하가 있을 수 있습니다.

assign(): 컨슈머가 특정 파티션을 직접 지정하여 소비합니다. 컨슈머 그룹 없이도 사용할 수 있으며, 리밸런싱이 발생하지 않습니다.

• 장점: 파티션을 직접 제어할 수 있고, 자동 리밸런싱 없이 특정 파티션을 소비할 수 있습니다.

• 단점: 여러 컨슈머가 있을 때 파티션을 직접 나누어 할당해야 하므로 관리가 복잡할 수 있습니다.

---

assign() 메서드의 장점

1. 리밸런싱 없이 안정적인 데이터 소비: assign()을 사용하면 리밸런싱이 발생하지 않기 때문에 컨슈머가 할당된 파티션에서 안정적으로 데이터를 소비할 수 있습니다. 이는 리밸런싱으로 인한 성능 저하를 피할 수 있다는 장점이 있습니다.

2. 특정 파티션에 대한 제어: 특정 파티션에서만 데이터를 읽어야 하는 경우, assign()을 사용하여 세밀한 파티션 제어가 가능합니다. 예를 들어, 특정 사용자의 데이터만 처리해야 하거나, 특정 이벤트 유형에 대한 데이터를 소비할 때 유용합니다.

3. 단일 컨슈머에서 여러 파티션 관리: 컨슈머 그룹을 사용하지 않거나, 여러 파티션을 단일 컨슈머에서 처리할 때 assign()을 통해 이를 제어할 수 있습니다. 여러 파티션에 대해 개별적으로 데이터를 소비할 수 있기 때문에, 컨슈머 그룹의 자동 분배 없이도 파티션 간의 데이터를 처리할 수 있습니다.

---

assign() 관련 설정 옵션

assign() 메서드를 사용할 때 중요한 몇 가지 설정 옵션이 있습니다.

• auto.offset.reset: 컨슈머가 처음 시작되거나 오프셋이 존재하지 않을 때, 어디서부터 데이터를 읽을지 결정하는 설정입니다. assign()을 사용할 때도 이 설정은 영향을 미칩니다.

  • earliest: 가장 오래된 메시지부터 읽습니다.

  • latest: 가장 최근의 메시지부터 읽습니다.

• enable.auto.commit: 자동으로 오프셋을 커밋할지 여부를 결정합니다. 수동으로 파티션을 할당하는 경우, 수동 커밋 방식을 사용하는 것이 좋습니다.

enable.auto.commit=false

이를 통해 컨슈머는 데이터를 정확히 처리한 후에만 오프셋을 커밋할 수 있습니다.

wakeup() 란?

wakeup() 메서드는 Kafka 컨슈머의 안전한 중단을 위한 메서드입니다. 일반적으로 Kafka 컨슈머는 poll() 메서드를 통해 블로킹 상태로 메시지를 가져옵니다. 이때 외부에서 컨슈머를 중단해야 할 상황이 발생하면, 바로 종료할 수 없고 현재 진행 중인 poll()이 끝날 때까지 기다려야 하는 문제가 발생할 수 있습니다.

이때 wakeup() 메서드를 호출하면, 현재 진행 중인 poll()을 즉시 중단시키고WakeupException을 발생시킵니다. 이 예외를 처리하여 컨슈머를 안전하게 종료할 수 있습니다.

wakeup() 메서드는 외부 신호(예: 애플리케이션 종료 시그널)를 받거나 컨슈머를 종료해야 할 때 사용됩니다. 예를 들어, 스레드가 Kafka 컨슈머를 실행 중일 때, 별도의 스레드에서 wakeup()을 호출하여 컨슈머 스레드를 중단시킬 수 있습니다.

아래는 wakeup()을 사용하여 Kafka 컨슈머를 안전하게 종료하는 예시입니다:

kotlin코드 복사val consumer = KafkaConsumer<String, String>(ClientConfiguration().config())
consumer.subscribe(listOf("my-topic"))

val thread = Thread {
    try {
        while (true) {
            val records = consumer.poll(Duration.ofMillis(100))
            records.forEach { record ->
                println("Consumed message: ${record.value()} from partition ${record.partition()}")
            }
        }
    } catch (e: WakeupException) {
        println("Consumer woken up and exiting...")
        // 여기서 스레드를 종료하거나 종료 처리 로직 실행
    } finally {
        consumer.close()  // 컨슈머 리소스 해제
    }
}

thread.start()

// 애플리케이션 종료 시 또는 외부 신호를 받을 때 호출
Runtime.getRuntime().addShutdownHook(Thread {
    consumer.wakeup()  // Wakeup 호출로 안전하게 poll 중단
    thread.join()       // 컨슈머 스레드 종료 대기
})

• Kafka 컨슈머는 별도의 스레드에서 실행되며, 데이터를 지속적으로 소비합니다.

• 애플리케이션이 종료되면 wakeup() 메서드가 호출되어 컨슈머의 poll()을 중단하고, WakeupException을 발생시킵니다.

• finally 블록에서 consumer.close()를 호출하여 리소스를 안전하게 해제합니다.

---

wakeup()의 주요 역할

1. 블록된 poll() 중단: Kafka 컨슈머가 poll() 메서드를 호출해 블로킹 상태로 데이터를 기다리는 동안, wakeup()을 호출하면 poll()이 즉시 중단되고 WakeupException이 발생합니다. 이를 통해 컨슈머는 즉시 종료될 수 있습니다.

2. 스레드 안전성: wakeup()은 스레드 안전합니다. 즉, 다른 스레드에서 호출해도 안전하게 실행됩니다. 이를 통해 메인 스레드에서 실행 중인 컨슈머 스레드를 안전하게 중단할 수 있습니다.

3. 컨슈머의 안전한 종료: Kafka 컨슈머는 보통 데이터를 지속적으로 소비하기 때문에, 애플리케이션 종료 시 안전하게 종료하는 것이 중요합니다. wakeup()은 poll() 중단 및 예외 처리를 통해 컨슈머 리소스를 안전하게 정리할 수 있도록 돕습니다.

---

wakeup()과 close()의 차이점

• wakeup(): 즉시 poll()을 중단하고 WakeupException을 발생시킵니다. 이를 통해 컨슈머 스레드를 안전하게 중단할 수 있습니다.

• close(): 컨슈머를 완전히 종료하고 모든 리소스를 해제합니다. wakeup()으로 poll()을 중단한 후, 컨슈머를 완전히 종료하려면 반드시 close()를 호출해야 합니다.

따라서 wakeup()은 현재 진행 중인 작업을 중단하는 역할을 하고, 그 후에 close()를 호출하여 컨슈머 리소스를 해제하는 것이 일반적인 패턴입니다.

Fetcher ?

Fetcher는 Kafka 컨슈머 내부에서 메시지를 가져오는 역할을 담당하는 중요한 구성 요소입니다.

Fetcher는 파티션의 리더 브로커로부터 메시지를 가져오는 작업을 수행하며, 여러 파티션에서 데이터를 병렬로 가져오는 것을 효율적으로 처리합니다.

• 메시지 가져오기: Fetcher는 컨슈머가 구독한 토픽의 각 파티션의 리더 브로커에 요청을 보내 메시지를 가져옵니다. 이는 각 브로커가 관리하는 파티션에서 메시지를 비동기로 가져오는 방식으로 이루어집니다.

• Fetch 요청 관리: Fetcher는 각 파티션에 대해 fetch 요청을 보내고, 오프셋을 기준으로 메시지를 가져옵니다. 가져온 메시지는 컨슈머가 처리할 수 있도록 제공됩니다.

• 데이터 일관성 유지: Fetcher는 오프셋을 기반으로 메시지를 읽어오며, Kafka의 일관성 모델을 따릅니다. 이를 통해, 데이터가 정확하게 처리되고, 메시지의 순서가 유지될 수 있도록 합니다.

• 페칭 성능 최적화: Kafka 컨슈머는 한 번에 여러 파티션에서 메시지를 가져오기 위해 최적화된 병렬 처리 구조를 사용합니다. Fetcher는 이 과정에서 최적의 성능을 보장하기 위해, 배치 크기, 네트워크 상태 등의 요소를 고려하여 메시지를 가져옵니다.

동작 원리

토픽 구독 및 파티션 할당: Kafka 컨슈머는 토픽을 구독하고, 해당 토픽의 파티션을 할당받습니다. 컨슈머 그룹 내에서 파티션은 자동으로 분배됩니다.

Fetcher를 통한 메시지 가져오기

Fetcher는 컨슈머가 할당받은 파티션에서 오프셋을 기준으로 메시지를 가져오는 역할을 수행합니다.

fetch 요청이 파티션의 리더 브로커로 보내지며, Fetcher는 브로커로부터 해당 오프셋 이후의 메시지를 가져옵니다.

오프셋 관리 및 메시지 처리

Fetcher가 가져온 메시지를 컨슈머는 오프셋을 기준으로 순차적으로 처리합니다.

메시지가 처리되면, 컨슈머는 오프셋을 커밋하여 처리된 메시지의 위치를 기록합니다.

리밸런싱과 파티션 재할당

컨슈머 그룹에서 리밸런싱이 발생할 경우, Fetcher는 새로운 파티션에 대해 fetch 요청을 다시 설정하여 데이터를 이어서 가져옵니다.