KangHun-Lee
commented
2 weeks ago 컨슈머 그룹과 파티션 리밸런스
컨슈머 그룹의 개념
- 컨슈머 그룹(Consumer Group)은 메시지를 병렬로 처리하기 위해 설계된 구조입니다.
- 같은 그룹에 속한 컨슈머들은 서로 다른 파티션을 할당받아 독립적으로 메시지를 처리합니다.
- 예를 들어, 3개의 파티션이 있는 토픽과 3명의 컨슈머가 있을 때, 각 컨슈머는 하나의 파티션을 할당받아 메시지를 병렬로 처리하게 됩니다.
파티션 할당 방식
- Kafka는 각 파티션을 컨슈머 그룹 내의 컨슈머들에게 자동으로 할당합니다.
- 만약 한 컨슈머가 추가되거나 제거되면, Kafka는 리밸런스를 통해 파티션을 다시 배분합니다.
- 예를 들어, 4개의 파티션과 2명의 컨슈머가 있을 경우, 각 컨슈머는 2개의 파티션을 할당받아 처리하게 됩니다.
컨슈머가 파티션의 갯수보다 많을 경우 유휴 컨슈머가 생기므로 주의 필요
파티션 리밸런스
리밸런스 개념
- 리밸런스는 컨슈머 그룹에 새로운 컨슈머가 추가되거나 기존 컨슈머가 제거될 때 파티션 할당을 재조정하는 과정입니다.
- 이 과정에서 일시적으로 메시지 처리가 중단될 수 있습니다.
조급한 리밸런스
- 이 방식은 메시지 처리 중단이 발생할 수 있습니다.
협력적 리밸런스
- 협력적 리밸런스(cooperative rebalance)는 메시지 처리 중단을 최소화하기 위한 방법입니다.
- 이 방식에서는 컨슈머들이 서서히 리밸런스에 참여하여 전체적인 리밸런스 과정이 보다 매끄럽게 진행되도록 합니다.
- 예를 들어, 새로운 컨슈머가 추가되면 기존 컨슈머와의 협력을 통해 리밸런스가 이루어지며, 이 과정에서 메시지 처리가 중단되지 않도록 합니다.
정적 그룹 멤버십
정적 그룹 멤버십의 필요성
- 정적 그룹 멤버십(Static Membership)은 컨슈머 그룹에서 리밸런스의 부하를 줄이기 위한 방법입니다.
- 이를 통해 동일한 컨슈머가 여러 번 그룹에 가입할 때 발생하는 리밸런스를 방지할 수 있습니다.
- 이 기능을 사용하면, 동일한 컨슈머가 재연결될 때 기존의 파티션 할당을 유지할 수 있어, 리밸런스가 필요 없게 됩니다.
리밸런스 이후의 처리
- 리밸런스가 발생한 이후에도 각 컨슈머는 할당받은 파티션을 올바르게 처리해야 합니다.
- 예를 들어, 컨슈머가 종료된 후 다시 시작되면, 이전에 처리 중이던 파티션의 위치를 기억하고 다시 처리할 수 있어야 합니다.
KafkaConsumer 생성 및 토픽 구독
KafkaConsumer 인스턴스 생성
- KafkaConsumer 인스턴스를 생성할 때는 부트스트랩 서버, 키, 디시리얼라이저 등의 기본 설정이 필요합니다.
- 이러한 설정을 통해 KafkaConsumer는 특정 Kafka 클러스터에 연결됩니다.
예시 코드
Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);토픽 구독
subscribe메소드를 사용하여 특정 토픽에 대해 구독할 수 있습니다.- 하나 이상의 토픽을 동시에 구독할 수 있으며, KafkaConsumer는 이 토픽들에서 메시지를 읽어들입니다.
예시 코드
consumer.subscribe(Collections.singletonList("customerCountries"));폴링 루프
폴링 루프 소개
- KafkaConsumer는 메시지를 지속적으로 가져오기 위해 폴링 루프(polling loop)를 사용합니다.
- 이 루프는 일정 시간 동안 메시지를 가져오고, 가져온 메시지를 처리합니다.
poll()메소드를 사용하여 메시지를 가져오며, 이 메소드는 설정된 타임아웃 내에 새로운 데이터를 반환합니다.
데이터 처리 과정
- 폴링 루프에서 가져온 메시지는 처리 로직을 거치게 됩니다.
- 메시지는 키와 값의 형태로 받아지며, 이를 적절히 처리한 후 필요한 경우 오프셋을 커밋(commit)하여 메시지가 중복 처리되지 않도록 합니다.
예시 코드
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
consumer.commitAsync();
}스레드 안정성
- KafkaConsumer는 스레드 안전하지 않으므로, 여러 스레드에서 동시에 접근할 경우 예기치 않은 동작이 발생할 수 있습니다.
- 이를 방지하기 위해 각 스레드는 별도의 KafkaConsumer 인스턴스를 사용해야 합니다.
- 또한, 컨슈머의 메소드 호출은 단일 스레드에서만 이루어져야 합니다.
- 컨슈머 로직을 자체적인 객체로 감싼 다음 자바의
ExecutorService를 사용하여 각자의 컨슈머를 가지는 다수의 스레드에서 실행하는 방법도 있습니다. - 이벤트를 받아서 큐에 넣는 컨슈머 하나와 이 큐에서 이벤트를 꺼내서 처리하는 여러 개의 워커 스레드를 사용하는 방법도 있습니다.
컨슈머 설정 옵션
중요 설정 옵션
1. auto.offset.reset
- 설명: 이 설정은 컨슈머가 처음으로 파티션을 읽거나, 기존 오프셋이 유효하지 않은 경우 데이터를 어디서부터 읽기 시작할지를 결정합니다.
- 중요성:
- 새롭게 추가된 컨슈머나 오프셋을 찾을 수 없는 상황에서 메시지 소비의 시작 위치를 결정하기 때문에, 데이터의 일관성과 처리 흐름에 큰 영향을 미칩니다.
- 예를 들어, "earliest"로 설정하면 컨슈머는 가능한 가장 이른 시점부터 메시지를 읽기 시작하여, 데이터 유실을 방지할 수 있습니다.
2. enable.auto.commit
- 설명: 이 설정은 컨슈머가 자동으로 오프셋을 커밋할지 여부를 결정합니다.
- 중요성:
- 자동 커밋을 사용하면 컨슈머가 주기적으로 오프셋을 커밋하여, 메시지가 중복 처리되는 것을 방지할 수 있습니다.
- 하지만 자동 커밋을 사용하면 메시지 처리 실패 시에도 오프셋이 커밋될 수 있어 데이터 손실의 위험이 있습니다.
- 따라서, 자동 커밋을 사용하지 않고 수동 커밋을 통해 메시지 처리 후에 오프셋을 커밋하는 것이 더 안전할 수 있습니다.
3. fetch.min.bytes 및 fetch.max.wait.ms
- 설명: 이 설정들은 컨슈머가 브로커로부터 데이터를 가져올 때의 최소 데이터 크기와 대기 시간을 조정합니다.
- 중요성:
- 이 두 설정은 네트워크 효율성과 지연 시간 간의 균형을 맞추는 데 중요합니다.
fetch.min.bytes가 높으면 네트워크 오버헤드를 줄일 수 있지만, 데이터 도착에 시간이 걸릴 수 있습니다.fetch.max.wait.ms는 이 대기 시간을 제한하여 지연을 최소화합니다.
4. max.poll.records
- 설명: 이 설정은 한 번의
poll()호출에서 가져올 수 있는 최대 레코드 수를 지정합니다. - 중요성:
- 이 설정은 컨슈머의 메모리 사용량과 메시지 처리량을 직접적으로 제어합니다.
- 너무 많은 레코드를 가져오면 메모리 사용량이 급증할 수 있으며, 너무 적은 레코드를 가져오면 처리 효율이 떨어질 수 있습니다.
5. session.timeout.ms 및 heartbeat.interval.ms
- 설명: 이 설정들은 컨슈머 그룹에서 컨슈머가 살아있는지 확인하기 위한 타임아웃과 하트비트 전송 간격을 설정합니다.
- 중요성:
- 이 설정들은 컨슈머 그룹의 안정성과 리밸런싱에 큰 영향을 미칩니다.
- 타임아웃이 너무 짧으면 컨슈머가 리밸런싱에 의해 자주 제거될 수 있고, 너무 길면 컨슈머 장애 감지가 지연될 수 있습니다.
fetch.min.bytes
- 설명: 컨슈머가 브로커에서 데이터를 가져오기 전에 기다릴 최소 데이터 크기를 지정합니다. 설정된 크기의 데이터가 모일 때까지 요청을 대기시킵니다.
- 기본값: 1바이트
- 사용 예시: 데이터량이 적을 때 네트워크 오버헤드를 줄이기 위해 사용합니다.
fetch.max.wait.ms
- 설명: 컨슈머가 데이터를 가져오기 위해
fetch.min.bytes를 만족하지 못할 경우 최대 대기 시간을 설정합니다. - 기본값: 500ms
- 사용 예시: 데이터를 더 빠르게 가져오고자 할 때 대기 시간을 조정합니다.
fetch.max.bytes
- 설명: 컨슈머가 한 번에 가져올 수 있는 최대 데이터 바이트 수를 지정합니다.
- 기본값: 50MB
- 사용 예시: 큰 메시지나 많은 양의 데이터를 처리할 때 유용합니다.
max.poll.records
- 설명: 한 번의
poll()호출에서 가져올 수 있는 최대 레코드 수를 지정합니다. - 기본값: 500
- 사용 예시: 처리할 레코드 수를 제한하여 메모리 사용량을 조절합니다.
max.partition.fetch.bytes
- 설명: 파티션당 한 번에 가져올 수 있는 최대 바이트 수를 설정합니다.
- 기본값: 1MB
- 사용 예시: 파티션에서 많은 데이터를 동시에 가져올 때 사용합니다.
session.timeout.ms 그리고 heartbeat.interval.ms
- 설명: 컨슈머가 브로커에 "정상 작동" 신호를 보내는 시간 간격과 세션 종료까지의 시간을 설정합니다.
- 기본값:
session.timeout.ms는 10초,heartbeat.interval.ms는 3초 - 사용 예시: 컨슈머 그룹 내에서 장애 감지를 위한 설정입니다.
max.poll.interval.ms
- 설명: 컨슈머가 데이터를 가져오기 위한 최대 대기 시간을 설정합니다. 이 시간 내에
poll()이 호출되지 않으면 컨슈머는 죽은 것으로 간주됩니다. - 기본값: 5분
- 사용 예시: 긴 작업을 수행하는 경우 이 값을 늘려줍니다.
default.api.timeout.ms
- 설명: 모든 API 호출의 기본 타임아웃을 설정합니다.
- 기본값: 60초
- 사용 예시: 네트워크 지연이 길거나 처리 시간이 긴 작업에서 타임아웃 조정에 사용합니다.
request.timeout.ms
- 설명: 컨슈머 요청의 타임아웃 시간을 설정합니다.
- 기본값: 30초
- 사용 예시: 네트워크 환경이 불안정할 때 타임아웃 시간을 늘리거나 줄일 수 있습니다.
auto.offset.reset
- 설명: 오프셋이 존재하지 않거나 유효하지 않을 경우의 동작을 정의합니다.
- 옵션: "earliest", "latest", "none"
- 사용 예시: 새로 시작한 컨슈머가 데이터를 어디서부터 읽기 시작할지를 결정합니다.
enable.auto.commit
- 설명: 컨슈머가 자동으로 오프셋을 커밋할지 여부를 결정합니다.
- 기본값: true
- 사용 예시: 자동 커밋을 통해 컨슈머의 오프셋 관리를 간소화합니다.
partition.assignment.strategy
- 설명: 파티션 할당 전략을 설정합니다.
- 옵션: "Range", "RoundRobin", "Sticky"
- 사용 예시: 컨슈머 그룹 내에서 파티션 할당 방식을 조정합니다.
Range: 토픽별로 파티션을 나누어 컨슈머에게 할당합니다. 이 방법은 특정 컨슈머가 여러 파티션을 독점하게 될 가능성이 있습니다. RoundRobin: 모든 컨슈머에게 파티션을 순서대로 할당합니다. Sticky: 파티션의 이동을 최소화하며, 이전 할당 상태를 최대한 유지하려고 합니다. Cooperative Sticky: Sticky 할당 전략과 유사하지만, 컨슈머가 리밸런스에 참여하는 방식이 다릅니다.
client.id
- 설명: 클라이언트를 식별하는 데 사용되는 ID를 설정합니다.
- 사용 예시: 모니터링과 로깅에 사용됩니다.
client.rack
- 설명: 클라이언트가 위치한 랙 정보를 설정하여 데이터 전송 최적화를 돕습니다.
- 사용 예시: 특정 랙에 있는 컨슈머에게 데이터를 우선적으로 전송하도록 설정합니다.
group.instance.id
- 설명: 정적 멤버십을 위한 그룹 인스턴스 ID를 설정합니다.
- 사용 예시: 컨슈머 그룹 내에서 특정 멤버를 유지하려는 경우 사용합니다.
receive.buffer.bytes, send.buffer.bytes
- 설명: 네트워크 전송 시 사용되는 버퍼 크기를 설정합니다.
- 사용 예시: 전송 속도와 메모리 사용을 최적화하기 위해 조정합니다.
offsets.retention.minutes
- 설명: 오프셋 보관 기간을 분 단위로 설정합니다.
- 사용 예시: 컨슈머 그룹 내의 오프셋을 얼마나 오래 보관할지를 결정합니다.
오프셋과 커밋
- 오프셋: Kafka에서 각 메시지가 차지하는 고유한 위치를 의미합니다. 각 파티션에서 메시지는 고유한 오프셋을 가집니다.
- 커밋: Kafka Consumer가 특정 오프셋까지의 메시지를 성공적으로 처리했음을 Kafka에 알리는 행위입니다. 이를 통해 동일 메시지가 중복 처리되지 않도록 보장합니다.
자동 커밋
- 설명:
enable.auto.commit을 true로 설정하면, Kafka Consumer는 일정 간격으로 자동으로 오프셋을 커밋합니다. - 장점: 설정이 간단하고, 자동으로 오프셋 관리가 됩니다.
- 단점: 메시지 처리 실패 시에도 오프셋이 커밋될 수 있어, 데이터 유실 가능성이 있습니다.
현재 오프셋 커밋하기
- 설명: 자동 커밋 대신 수동으로 오프셋을 커밋할 수 있습니다. 이를 통해 메시지를 안전하게 처리한 후에만 오프셋을 커밋할 수 있습니다.
비동기적 커밋
- 설명:
commitAsync()를 사용하여 비동기적으로 오프셋을 커밋할 수 있습니다. 이는 커밋 요청을 비동기적으로 처리하여 더 빠른 성능을 제공합니다.
동기적 커밋과 비동기적 커밋을 함께 사용하기
- 설명: 안전성을 높이기 위해, 정기적으로 동기적 커밋과 비동기적 커밋을 함께 사용할 수 있습니다.
특정 오프셋 커밋하기
- 설명: 특정 파티션의 특정 오프셋을 커밋할 수 있습니다. 이를 통해 커밋 전략을 보다 유연하게 관리할 수 있습니다.
리밸런스 리스너
- 설명: 리밸런스는 Kafka에서 컨슈머 그룹의 파티션 할당이 변경될 때 발생합니다. 리밸런스 리스너를 구현하면, 리밸런스가 일어나기 전과 후에 특정 작업을 수행할 수 있습니다.
- 사용 예시: 리밸런스가 발생하기 전에 현재 오프셋을 커밋하거나, 리밸런스가 완료된 후 새로운 파티션 할당을 처리할 수 있습니다.
특정 오프셋의 레코드 읽어오기
- 설명: 특정 오프셋에서부터 레코드를 읽어오는 방법입니다. 이 방법은 데이터 복구나 특정 지점부터 다시 데이터를 처리할 때 유용합니다.
- 예시 코드:
consumer.seek(new TopicPartition("topic", partition), offset);
폴링 루프를 벗어나는 방법
- 설명: 폴링 루프는 Kafka Consumer가 데이터를 지속적으로 가져오는 루프입니다. 하지만 이 루프를 안전하게 벗어나려면 특정 방법을 사용해야 합니다.
- 방법: ShutdownHook을 사용하여 종료 시점을 잡거나,
WakeupException을 활용하여 안전하게 폴링 루프를 종료할 수 있습니다.
디시리얼라이저
- 설명: Kafka는 메시지를 바이트 배열로 저장하므로, 이를 적절한 객체로 변환하기 위해 디시리얼라이저를 사용합니다. 기본적으로 제공되는
StringDeserializer,IntegerDeserializer외에, 사용자 정의 디시리얼라이저를 구현할 수 있습니다. - 사용 예시:
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
커스텀 디시리얼라이저
- 설명: Kafka Consumer에서 특정 요구 사항에 맞는 커스텀 디시리얼라이저를 구현하여 사용합니다.
- 예시 코드:
public class CustomerDeserializer implements Deserializer<Customer> { @Override public Customer deserialize(String topic, byte[] data) { // 데이터 변환 로직 } }
Avro 디시리얼라이저 사용하기
- 설명: Avro와 같은 포맷을 사용하여 더 복잡한 메시지 구조를 처리할 수 있습니다. 이를 위해 Avro 디시리얼라이저를 사용합니다.
- 예시 코드:
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
독립 실행 컨슈머
- 설명: 컨슈머 그룹 없이 독립적으로 실행되는 컨슈머입니다. 특정 파티션의 메시지를 개별적으로 처리할 때 사용됩니다.
- 사용 예시:
subscribe대신assign메소드를 사용하여 특정 파티션을 할당받아 데이터를 처리합니다.
CHOICORE
zinokim
hubtwork
iamzin
kimsunhak
Chapter Ownership : @KangHun-Lee