[Kafka] 트랜잭션, exactly-once

Kafka Transaction?

스트림 처리 애플리케이션을 위해 도입된 기능이다. exactly-once 을 보장해준다.

동작

• 트랜잭션 프로듀서를 사용할 경우, 트랜잭션 코디네이터가 트랜잭션를 관리하게 된다.
• 출력 토픽에 레코드 저장, 오프셋 증가 2가지 작업이 동시에 커밋되는 것을 보장한다. (Atomic multipartition write)

트랜잭션이 해결할 수 있는 문제들

1. 앱 크래시로 인한 중복 메시지 문제

• 앱이 토픽에 메시지를 쌓았는데 커밋은 못한 경우 중복 메시지가 쌓일 수 있음.
• Atomic multi partition write 보장으로 해결

2. 브로커 응답 문제로 인한 인한 중복 메시지 문제

• 브로커가 커밋은 성공했는데 성공 응답에 실패한 경우, 프로듀서는 재요청을 보내며 중복 메시지가 쌓일 수 있음.
• Atomic multi partition write 보장으로 해결

3. 좀비 펜싱 중복 문제

• 이미 연결이 끊긴 컨슈머가 자신이 죽은 상태를 인지하지 못하는 상태를 좀비 상태라고 부르는데, 이때 보낸 메시지가 토픽에 쌓여 중복 메시지가 쌓일 수 있음
• 트랜잭션을 사용하면 initTransactions 할 때 에포크를 증가시키기 때문에, 이것보다 작은 에포크를 지닌 프로듀서의 요청은 무시됨

트랜잭션이 해결할 수 없는 문제들

1. 사이드 이펙트

• 카프카 레코드 컨슘이 1회 보장될 뿐이다. → 스트림 앱에서 메일을 보내는 경우 메일은 2번 간다.

적용 방법

1. 클러스터
   1. transaction.state.log.replication.factor
      1. 기본 3
      2. 트랜잭션 상태 로그 토픽의 복제 인수(replication factor)를 지정
      3. 트랜잭션 상태 로그는 트랜잭션 상태를 저장하는데 사용되는 내부 토픽. 이 값을 높게 설정하면 트랜잭션 상태 로그가 더 많은 브로커에 복제되어 장애 발생 시 데이터 손실 가능성을 줄일 수 있음
   2. transaction.state.log.min.isr
      1. 기본 2
      2. 트랜잭션 상태 로그 토픽의 최소 ISR(In-Sync Replica) 수를 지정
      3. ISR은 복제된 데이터의 최신 복사본을 유지하는 브로커들의 집합
      4. 값을 높게 설정하면 트랜잭션의 내구성을 더욱 보장 있음
         1. 예를 들어, 이 값을 2로 설정하면 트랜잭션 상태 로그를 처리할 때 최소 두 개의 브로커가 최신 상태를 유지해야 함.
2. 프로듀서
   1. 트랜잭션 프로듀서 설정 (transactional_id 설정)
   2. 트랜잭션 처리 관련 메서드 사용
      1. producer.initTransactions()
      2. producer.beginTransaction()
      3. producer.sendOffsetToTransaction()
      4. producer.commitTransaction()
      5. producer.abortTransaction()
3. 컨슈머
   1. isolation.level 옵션 read_committed 로 설정
   2. enable_auto_commit_config 옵션 false 설정

성능

• 트랜잭션은 프로듀서에 약간의 오버헤드를 발생시킨다
  • 트랜잭션 ID 등록 API 요청은 최초 1회 발생하기 때문에 큰 영향은 없다.
  • 트랜잭션 초기화, 커밋 요청은 동기적으로 작동한다.
    • 성공적으로 완료되거나 ,실패하거나, 타임아웃되기 전까지는 어떤 데이터도 전송되지 않는다.
• 트랜잭션은 컨슈머에 직접적인 오버헤드를 발생시키지는 않는다.
  • 커밋되지 않은 메시지를 아예 읽어오지 않기 때문에 오버헤드가 없다.
  • 다만 트랜잭션 완료 처리가 늦어질 경우, 메시지가 늦게 온다.

모니터링

• LSO 값을 모니터링 필요
  • 마지막 트랜잭션 처리 커밋 offset 과 최신 offset 간격을 나타낸다.
  • 이 값이 무한히 증가하면 트랜잭션이 멈춰있을 가능성이 높다.
  • kafka_cluster_partition_laststableoffsetlag