Kafka Transaction

Transaction

• 카프카 트랜잭션은 주로 비동기 데이터 스트림에서 읽고 쓰는 패턴을 가진 애플리케이션을 위해 설계되었다.
• 초기에는 스트림 처리 애플리케이션이 부정확한 처리를 허용했음. (웹 페이지 조회수, 좋아요 수)
• 그러나 카프카에 대한 인기가 증가하면서 정확한 처리에 대한 요구사항들이 생김(인출, 입금 등)
• idempotent Producer는 다중 파티션에 대한 보증은 할 수 없다.

Transactional Semantics

• Atomic multi-partition writes
  • 트랜잭션은 다수의 토픽과 파티션에 대한 원자적인 쓰기를 가능하게 한다
  • 트랜잭션에 포함된 메시지들은 전체 성공하거나 아니거나 둘 중에 하나
• Zombie fencing
  • 최초 프로듀서가 시작할 때 카프카 클러스터에 transactional.id라고 불리는 유니크한 식별자를 등록한다. 브로커는 주어진 transactional.id를 이용해 오픈된 트랜잭션이 있는지 확인한다. 또한 클러스터는 연관된 epoch를 증가시키는데, 에포크는 모든 transactional.id에 대해 저장되는 내부 메타데이터이다. 에포크가 증가되면 동일한 transactional.id와 이전 에포크를 가진 프로듀서들은 좀비로 간주되며 차단한다.
• Reading Transactional Messages
  • 컨슈머는 오직 커밋된 메시지만 전달받는다.
  • 컨슈머에서 소비하는 메시지가 어떤 트랜잭션의 일부인지는 알 수 없다.

 

Transaction Key Concepts

Transaction Coordinator

• 컨슈머 그룹 코디네이터와 비슷
• 각 프로듀서에게는 트랜잭션 코디네이터가 할당됨.
• 트랜잭션에 코디네이터는 해당 프로듀서들의 모든 로직을 관리한다

Transaction Log

• 컨슈머 오프셋 토픽과 비슷하다
• 트랜잭션 레코드는 트랜잭션 코디네이터의 상태 스토어 개념이다.
• 모든 트랜잭션 레코드는 영속되고 복제된다.

Controller message

• 유저 토픽에 쓰이는 특별한 메시지(or 레코드)이며 유저에게는 절대 노출되지 않는다.
• 예를 들어 컨슈머가 이전에 가져간 메시지가 원자적으로 커밋된 것인지, 아닌지를 브로커가 알려주는데 사용할 수 있다.

 

Transaction Date Flow

1. 애플리케이션에서 랜덤한 브로커로부터 트랜잭션 코디네이터가 존재하는 브로커를 찾아낸다.
2. 트랜잭션 코디네이터에게 transactional.id를 등록한다. 
   • transactional.id가 명시되지 않았다면 PID를 새로 발급하고, 해당 세션이 트랜잭션을 사용할 수 있지만 이후 해당 애플리케이션이 재시작되어도 이전 트랜잭션의 존재를 알 수 없다.
   • transactiona.id가 명시되어 있다면 PID를 대응시켜 해당 값으로 트랜잭션 로그에 저장한다. 아마 이전에 등록된 적 있는 transactiona.id라면 동일한 PID가 대응돼서 저장될 것이고, 아니라면 새로 PID가 발급된다. 이전 애플리케이션에서 완료하지 못한 채 남겨진 트랜잭션을 재생하거나 중단 시킬 수 있다. 좀비 인스턴스를 막기 위해 epoch값을 범프(증가)시킨다.
3. 트랜잭션 시작
   • 애플리케이션(=프로듀서)에서 beginTransaction() 메서드를 호출해서 내부적으로 트랜잭션이 시작되었음을 알린다. 하지만 이 타이밍에 트랜잭션 코디네이터에게는 트랜잭션 시작을 전달하지 않으므로, 첫 번째 레코드가 쌓이기 전까지는 시작되었는지는 알 수 없다.
4. 메시지 전송
   • 어떤 토픽의 파티션에 메시지를 쓰는지 먼저 트랜잭션 코디네이터에 통보한다. 이 정보는 후에 코디네이터가 커밋 혹은 중단시 해당 파티션에 컨트롤 메시지를 남기기 위해 사용한다.
   • 위 타이밍에 트랜잭션 코디네이터는 트랜잭션 타이머를 시작시킨다 = max.transction.timeout.ms
   • 프로듀서가 해당 파티션에 PID, epoch, sequence number 메타데이터 포함된 메시지를 전달한다.
     • 위 메타데이터는 실제 사용할 때는 사용자가 직접 세팅할 필요없다.
     • 메타데이터는 후에 컨슈머에서 커밋된 데이터인지 아닌지를 구분하는 과정에서 사용되고, 트랜잭션의 메시지 순서를 인식하기 위해 사용된다.
   • 만약 애플리케이션이 컨슈머를 포함하는 경우 특정 메시지를 읽어다는 것을 표시하기 위해 Consumer Coordinator에도 메시지를 보내야 하는데 해당 메시지 역시 트랜잭션 범위이므로 애플리케이션에서 sendToOffsetsTransaction을 이용할 수 있다. 그러면 트랜잭션 라이프 사이클을 사이에 껴들 수 있다.
5. 커밋 혹은 중단
   • 트랜잭션 코디네이터에게 커밋 혹은 중단 요청을 날린다.
   • 트랜잭션 코디네이터는 요청을 받으면 아래와 같은 단계를 거친다.
     • PREPARE_COMMIT 혹은 PREPARE_ABORT 메시지를 트랜잭션 로그에 먼저 쓴다
     • COMMIT 혹은 ABORT 마커를 트랜잭션 로그에 기록된 파티션들에게 쓸 수 있도록 각 브로커에 요청한다. 요청을 받은 브로커는 해당 파티션에 마커를 남긴다. 
     • 마커가 남겨지면 COMMITTED 혹은 ABORTED 메시지를 트랜잭션 로그에 쓴다.

 

Transaction Overviews

https://developer.confluent.io/courses/architecture/transactions/

Kafka Transactional Support: How It Enables Exactly-Once Semantics