4. 컨슈머(Consumer)의 모든 것: 메시지 소비 방식과 컨슈머 그룹

컨슈머와 컨슈머 그룹의 역할

카프카 컨슈머(Consumer)는 토픽에 저장된 메시지를 가져와서(consume) 소비하는 클라이언트 애플리케이션입니다. 하지만 카프카의 진정한 힘은 개별 컨슈머가 아닌, 여러 컨슈머가 협력하는 컨슈머 그룹(Consumer Group)에서 나옵니다.

• 컨슈머 그룹: 동일한 group.id를 공유하는 컨슈머들의 집합입니다. 그룹 내의 컨슈머들은 하나의 토픽을 병렬로 처리하기 위해 파티션들을 나누어 가집니다.
• 파티션 소유권 규칙: 컨슈머 그룹 내에서, 하나의 파티션은 단 하나의 컨슈머에 의해서만 소비될 수 있습니다. 이 규칙 덕분에 동일한 메시지가 그룹 내에서 중복으로 처리되는 일이 없습니다.
• 독립적인 소비: 서로 다른 컨슈머 그룹은 같은 토픽을 구독하더라도 서로에게 아무런 영향을 주지 않고 독립적으로 메시지를 소비할 수 있습니다. 각 그룹은 자신만의 오프셋을 관리합니다.

그룹 A에서는 2개의 컨슈머가 3개의 파티션을 나누어 처리하고, 그룹 B에서는 1개의 컨슈머가 모든 파티션을 처리합니다.

컨슈머 그룹의 마법, 리밸런싱(Rebalancing)

리밸런싱은 컨슈머 그룹 내의 파티션 소유권을 동적으로 재조정하는 과정입니다. 그룹에 새로운 컨슈머가 참여하거나 기존 컨슈머가 떠날 때 발생하며, 이를 통해 카프카는 높은 가용성과 확장성을 유지합니다.

하지만 리밸런싱은 공짜가 아닙니다. 리밸런싱이 진행되는 동안에는 그룹 내 모든 컨슈머가 메시지 처리를 일시적으로 중단합니다. 따라서 불필요하게 잦은 리밸런싱은 시스템 전체의 성능 저하를 초래할 수 있습니다.

"어디까지 읽었지?" - 오프셋 커밋(Offset Commit)

컨슈머는 자신이 그룹 내에서 특정 파티션의 메시지를 어디까지 처리했는지 그 위치(오프셋)를 카프카에 기록해야 합니다. 이를 '오프셋 커밋'이라고 합니다. 이 커밋 정보 덕분에 컨슈머에 장애가 발생했다가 복구되어도, 마지막으로 처리했던 위치부터 이어서 작업을 재개할 수 있습니다.

1. 자동 커밋 (enable.auto.commit=true)

가장 간편한 방식이지만, 데이터 처리 여부를 보장하지 않기 때문에 심각한 데이터 유실이나 중복을 유발할 수 있습니다. 이 때문에 중요한 데이터를 다루는 실무 환경에서는 절대 사용해서는 안 되는 옵션으로 간주됩니다. auto.commit.interval.ms(기본 5초)마다 컨슈머가 poll()로 가져간 메시지 묶음의 가장 마지막 오프셋을 처리 여부와 상관없이 자동으로 커밋합니다. 가장 간편한 방식이지만, 데이터 처리 여부를 보장하지 않기 때문에 심각한 데이터 유실이나 중복을 유발할 수 있습니다. 이 때문에 중요한 데이터를 다루는 실무 환경에서는 절대 사용해서는 안 되는 옵션으로 간주됩니다. auto.commit.interval.ms(기본 5초)마다 컨슈머가 poll()로 가져간 메시지 묶음의 가장 마지막 오프셋을 **처리 여부와 상관없이** 자동으로 커밋합니다.

2. 수동 커밋 (enable.auto.commit=false)

개발자가 코드에서 명시적으로 커밋 시점을 제어하는 방식으로, 데이터 처리의 정합성을 보장하기 위한 필수적인 선택입니다. 가장 일반적인 패턴은 "선(先)처리, 후(後)커밋" 입니다.

• 동기 커밋 (commitSync()): 브로커로부터 오프셋 커밋이 성공적으로 완료되었다는 응답을 받을 때까지 컨슈머가 대기(Blocking)합니다. 구현이 간단하고 직관적이며, 커밋이 실패하면 예외가 발생하므로 try-catch로 재시도 로직을 구현하기 용이합니다. 처리량에 약간의 영향을 줄 수 있습니다.
• 비동기 커밋 (commitAsync()): 컨슈머가 브로커한테 커밋 요청을 보낸 후 커밋 요청의 성공 혹은 실패 응답을 기다리지 않고 즉시 다음 로직을 수행합니다. 처리량이 높지만, 커밋 결과를 '콜백(Callback)'으로 받아 처리해야 하므로 구현이 더 복잡합니다.

  💡 콜백(Callback)이란?

  "이 작업이 끝나면, 그 결과를 가지고 이 함수를 실행해 줘"라고 시스템에 등록하는 함수입니다. commitAsync의 경우, 커밋 요청이 성공하든 실패하든 브로커로부터 응답이 왔을 때, 우리가 미리 전달해 둔 콜백 함수가 실행됩니다.

  이 콜백 함수 안에서 커밋 실패 시의 로깅이나 재시도 로직을 구현할 수 있습니다. 동기 커밋의 try-catch 구문과 달리, 에러 처리가 별도의 스레드에서 비동기적으로 일어나기 때문에 로직이 더 복잡해집니다.

  
  consumer.commitAsync((offsets, exception) -> {
      if (exception != null) {
          // 커밋 실패 시 처리 로직
          log.error("Failed to commit offsets {} with exception", offsets, exception);
      } else {
          // 커밋 성공 시 처리 로직 (보통은 비워둠)
          log.debug("Successfully committed offsets {}", offsets);
      }
  });
                  

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        // 1. 메시지를 가져와서
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

        // 2. 가장 중요한 비즈니스 로직을 먼저 처리합니다. (e.g., DB에 저장)
        processRecord(record);
    }
    try {
        // 3. 모든 로직이 성공적으로 끝난 후에, 마지막으로 오프셋을 커밋합니다.
        consumer.commitSync();
    } catch (CommitFailedException e) {
        // 커밋 실패 시 예외 처리 (e.g., 로깅, 재시도)
        log.error("Failed to commit offsets", e);
    }
}
        

주요 컨슈머 설정 파라미터