[Kafka] 고급 설정

📌 Kafka 운영 환경에서 필요한 고급 설정

기본 개념만으로도 Kafka를 사용할 수는 있지만, 실제 운영 환경에서는 더 많은 걸 알아야 한다!

🔗 기본 개념(필수!)

• 토픽, 파티션, 브로커, 컨슈머 그룹, 오프셋 이해
• Producer & Consumer 동작 방식
• 기본적인 명령어 (토픽 생성, 메시지 생산/소비, 오프셋 관리)

➡️ 이 정도만 알아도 로컬 환경에서 테스트하고 간단한 서비스에 적용할 수 있다!! 🚀

 

📌 1. 파티션 할당 전략 (Partition Assignment Strategy)

 

✅ 컨슈머 그룹 최적화

Kafka는 여러 개의 컨슈머가 동일한 토픽을 소비할 때, 파티션을 컨슈머들에게 자동으로 할당한다. 이때 파티션 할당 전략이 중요한 역할을 한다.

 ➡️ 운영 환경에서는 컨슈머 그룹의 부하 분산을 최적화하기 위해 상황에 맞는 할당 전략을 선택해야 한다.

 

✅ 주요 할당 전략

• RangeAssignor (기본값): 파티션을 연속적인 범위로 나누어 할당. 컨슈머 수와 파티션 수가 균등하지 않으면 일부 컨슈머가 더 많은 파티션을 가져갈 수 있음.
• RoundRobinAssignor: 모든 컨슈머에게 **순환 방식(라운드로빈)**으로 골고루 할당. 컨슈머 간 균등한 분배가 필요한 경우 적합.
• StickyAssignor: 파티션 재할당 시 기존 할당을 최대한 유지하여 리밸런싱 오버헤드 최소화.
• Custom Assignor: 특정 요구사항에 맞게 직접 구현 가능.

✅ 할당 전략 설정 방법

partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

 

⭐ 파티션의 개수

  ➡️ 파티션 & 브로커 개수 관계

• 파티션 개수는 브로커 개수와 무관하게 설정 가능!
  → 하지만 최적 성능을 위해 브로커 개수보다 크거나 같게 설정하는 게 일반적.
• 이유: 각 브로커가 최소한 하나의 파티션을 가져야 부하가 분산됨!
  → 파티션 개수가 브로커 개수보다 적으면 일부 브로커가 놀게 됨.

 

  ➡️ 추천: 파티션 개수 ≥ 브로커 개수 (보통 브로커 개수 × 2 추천)

 

👉 파티션 개수가 브로커 개수보다 적으면

🔍 남는 브로커는 언제 역할을 할까?

1️⃣ 장애 발생 시 대기 역할

• 리더 브로커가 죽으면 남은 브로커가 리더로 승격될 수도 있음
• 남는 브로커로 새로운 파티션이 배치될 수도 있음 (브로커 추가 시 자동 조정)

 

2️⃣ 다른 토픽의 파티션을 저장할 수도 있음

• 운영 환경에서는 보통 여러 개의 토픽이 있음
• 특정 토픽에는 역할이 없더라도 다른 토픽의 리더 또는 팔로워로 쓰일 수 있음

 

📌 2. 리플리케이션 & 장애 대비 (Replication & Fault Tolerance)

✅  Kafka는 데이터 유실을 방지하기 위해 리플리케이션(복제) 기능을 제공한다.

 ➡️ 운영 환경에서는 리플리케이션을 통해 데이터 유실을 방지하고, 장애 발생 시 빠른 복구가 가능하도록 설정해야 한다

⭐ 리플리케이션의 개수

  ➡️ 리플리케이션 & 브로커 개수 관계

• 최대 브로커 개수만큼 설정 가능. (이 때 브로커가 장애 나면, 새롭게 할당할 공간이 없어 복구가 어려움)
• 보통 브로커 개수 -1로 설정 (1개는 리더, 나머지는 팔로워).
• 모든 복제본은 서로 다른 브로커에 저장돼야 하므로,
  → replication-factor는 브로커 개수를 초과할 수 없음.

  ➡️  추천: replication-factor ≤ 브로커 개수 (보통 브로커 개수 - 1 추천)

 

📖 리플리케이션의 할당 예

토픽 A (브로커 4개, 파티션 3개, 복제 3개)

Kafka는 자동으로 브로커를 최대한 균등하게 분배

파티션	리더 브로커	팔로워 브로커 1	팔로워 브로커 2
P0	B1	B2	B3
P1	B2	B3	B4
P2	B3	B4	B1

✅ 구성 요소

• 리더 파티션: 클러스터에서 특정 브로커가 해당 파티션의 리더 역할을 담당.
• 팔로워 파티션: 리더를 복제하여 백업 역할을 수행.
• ISR (In-Sync Replicas): 리더와 동기화된 팔로워 목록. 장애 발생 시 새로운 리더를 선출하는 데 사용됨.

 

✅ 작동 방식

• 각 파티션은 하나의 리더(Leader)와 하나 이상의 팔로워(Follower)로 구성됨.
• 프로듀서(Producer)는 항상 리더 파티션으로 데이터를 보냄.
• 팔로워 파티션들은 리더의 데이터를 동기화(복제)하면서 대기.

 

✅ 장애 발생 시 복구 과정 

1️⃣ 리더 브로커 장애 발생 🚨 → ISR 내에서 새로운 리더 자동 선출.
2️⃣ 팔로워가 일정 시간 리더와 동기화되지 않으면 ISR에서 제거됨.
3️⃣ 복구된 브로커는 다시 ISR에 추가됨.

 

✅ 설정 방법

min.insync.replicas=2  # 또는 토픽 생성 시 replication-factor=2 # 최소한 2개의 복제본이 있어야 커밋 가능

unclean.leader.election.enable=false  # 손상된 데이터가 리더로 선출되는 것을 방지

 

 

 

📌 3. 레텐션 정책 & 성능 튜닝 (Retention Policy & Performance Tuning)

✅ Kafka는 오래된 데이터를 자동으로 삭제하거나 유지하는 레텐션 정책을 지원한다.

 ➡️ 운영 환경에서는 레텐션 설정을 통해 저장 공간을 관리하고, 성능 튜닝을 통해 처리 속도를 최적화해야 한다.

 

✅ 레텐션 정책 설정 방법

log.retention.ms=604800000  # 7일간 데이터 보관

log.retention.bytes=1073741824  # 1GB 초과 시 삭제

 

✅ 성능 튜닝 요소

• 배치 크기 조정: batch.size 값 조절로 프로듀서의 성능 향상.
• 압축 사용: compression.type=lz4 설정으로 데이터 전송량 절감.
• 메시지 크기 제한: message.max.bytes 설정으로 개별 메시지 크기 조정.

 

📌 4. Kafka Connect & Schema Registry (외부 시스템 연동)

✅ Kafka Connect를 사용하면 다른 시스템과 손쉽게 데이터를 연동할 수 있다.

 ➡️ 운영 환경에서는 Kafka Connect를 활용하여 다양한 시스템과 연동하고, Schema Registry로 데이터 일관성을 유지해야 한다.

 

✅ Kafka Connect의 역할

• DB, 클라우드, NoSQL, 파일 시스템 등 다양한 소스에서 Kafka로 데이터를 가져오거나 전송 가능.
• Sink Connector: Kafka → 외부 시스템 (예: Elasticsearch, HDFS, RDBMS)
• Source Connector: 외부 시스템 → Kafka (예: MySQL, PostgreSQL, MongoDB)

 

✅ Schema Registry 활용

• Kafka 메시지의 JSON, Avro, Protobuf 같은 데이터 형식을 관리.
• 스키마 변경 시 데이터 무결성을 유지하고, 데이터 손실을 방지.
• REST API를 통해 스키마를 조회하고 관리 가능.

 

📌 5. Kafka Streams & KSQL (실시간 데이터 처리)

✅ Kafka Streams: Kafka 데이터를 실시간으로 처리하는 강력한 스트리밍 라이브러리.
✅ KSQL: SQL 문법을 이용해 실시간 데이터 처리를 쉽게 구현할 수 있다.

 ➡️ 운영 환경에서는 Kafka Streams 또는 KSQL을 활용하여 실시간 데이터 처리를 최적화할 수 있다.

 

✅ Kafka Streams 주요 개념

• Stateful & Stateless 연산 지원
• Windowing 처리 가능 (시간 기반 데이터 처리)
• Exactly-once 처리 가능 (중복 메시지 방지)

 

✅ Kafka Streams 예제

KStream<String, String> stream = builder.stream("input-topic");

stream.filter((key, value) -> value.contains("important"))

      .to("output-topic");

 

✅ KSQL 주요 개념

• Kafka 토픽을 SQL처럼 조회하고 필터링 가능.
• 실시간 데이터 스트림을 조인, 집계, 변환 가능.

 

✅ KSQL 예제

SELECT * FROM transactions WHERE amount > 1000;

 

📌 6. Kafka 운영 환경에서의 오프셋 커밋 전략

✅  Kafka에서 오프셋(Offset) 은 컨슈머가 특정 토픽의 파티션에서 데이터를 어디까지 읽었는지를 나타내는 번호다.
 ➡️ 운영 환경에서는 데이터 유실 방지, 중복 방지, 정확한 한 번만 처리 (Exactly-Once) 보장을 위해 오프셋 커밋 전략이 중요함!

 

✅ 1. 오프셋 커밋 방식: 자동 vs 수동

오프셋을 언제, 어떻게 저장할지 결정하는 설정.

방식	설명	장점	단점
자동 커밋 (Auto Commit)	Kafka가 auto.commit.interval.ms(기본 5초)마다 자동으로 오프셋 저장	구현이 간단	메시지가 처리되기 전에 커밋될 수도 있음 (데이터 유실 가능)
수동 커밋 (Manual Commit)	컨슈머가 직접 오프셋을 커밋 (commitSync(), commitAsync())	데이터 유실 & 중복 최소화 가능	개발자가 직접 관리해야 함

✔ 운영 환경에서는 자동 커밋보다 수동 커밋을 쓰는 게 안전!
✔ commitSync()는 정확하지만 느리고, commitAsync()는 빠르지만 일부 실패 가능 → 둘을 조합해서 사용 가능!

 

📖 자동 커밋 비활성화

enable.auto.commit=false

 

📖 일반 Kafka 컨슈머 예제 (commitSync() 사용)

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, String> record : records) {
        processRecord(record); // 메시지 처리
    }

    consumer.commitSync(); // 처리 완료 후 오프셋 커밋
}

⭐ @KafkaListener 사용 시 수동 커밋

📖 Spring Kafka 예제 (ack.acknowledge() 사용)

@KafkaListener(topics = "my-topic", containerFactory = "myFactory")
public void listen(String message, Acknowledgment ack) {
    System.out.println("메시지 수신: " + message);
    ack.acknowledge();  // 메시지 정상 처리 후 오프셋 커밋
}

 

📖 KafkaListenerContainerFactory에서 ackMode 변경

위 코드가 제대로 동작하려면, 리스너 컨테이너의 ackMode를 MANUAL로 설정해야 한다.

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, DeliveryStartedEvent> kafkaListenerContainerFactory(
            ConsumerFactory<String, DeliveryStartedEvent> consumerFactory) {
    
        ConcurrentKafkaListenerContainerFactory<String, DeliveryStartedEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();

        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); // 수동 커밋 설정

        return factory;
    }
}

 

 

✅ 2. 오프셋 초기화 (Reset) - 메시지 유실 방지

컨슈머가 처음 시작하거나, 오프셋 정보가 사라졌을 때 Kafka가 어디서부터 읽을지 결정하는 설정.

운영 환경에서는 earliest를 설정해서 데이터 유실을 방지하는 게 좋음!

설정 값	동작 방식
earliest	가장 오래된 메시지부터 읽음 (데이터 유실 방지)
latest (기본값)	가장 최근 메시지부터 읽음 (이전 메시지는 무시됨)

📖 오프셋 초기화 설정

auto.offset.reset=earliest

 

 

✅ 3. Exactly-Once 보장 (중복 방지 & 트랜잭션 처리)

Kafka 기본 설정에서는 at-least-once 방식이라 중복 메시지 소비가 발생할 수 있다.
이를 방지하려면 Exactly-Once 처리를 위한 설정을 추가해야 함!

• idempotent producer (중복 방지 프로듀서) + transactional consumer (트랜잭셔널 컨슈머) 를 조합.
• read-process-write 패턴에서 소비한 메시지와 새로운 메시지를 하나의 트랜잭션으로 묶어 커밋함.

📖 프로듀서 측 설정 (중복 방지)

enable.idempotence=true
acks=all

 

📖 컨슈머 측 설정 (트랜잭션 활성화)

isolation.level=read_committed

 

📖 트랜잭션al 소비자 + 프로듀서 사용

Properties props = new Properties();
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transaction-id");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    producer.beginTransaction();
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
        producer.send(new ProducerRecord<>("new-topic", record.value()));
    }
   
    producer.commitTransaction(); // 프로듀서 트랜잭션 커밋
    consumer.commitSync(); // 컨슈머 오프셋 커밋
}