[개발생각] Kafka 이벤트와 트랜잭션에 대해서

트랜잭션과 비동기 프로세스의 처리에 대해서, 인지하고 있어야 할것이 있다.

 

가끔 카프카 이벤트와 트랜잭션에 대해서 혼돈이 일어날때가 있어서, 적어두고자 한다.

 

@Transactional

fun saveUser() {

    userRepository.save(User(id = "test"))

    kafkaEventPublisher.send(Event("SaveUser"))

}



@Transactional

@KafkaListener

fun listener(event:String) {

    val user = userRepository.findById("test")

    user?.let {

        println("user is not null")

    } ?: println("user is null")

}

 

이경우,  대충보면, listener함수에서 당연히 user is not null 이라고 출력이 될것으로 파악할수있다.

하지만, 결과는 user is null로 출력된다.

 

왜나면, saveUser에서 id=test인 user가 커밋되기 전에,

카프카 이벤트가 발생하고, listener로 이벤트가 들어온 시점에도 아직 saveUser가 커밋이 되지 않은 상태이기 때문에,

user를 조회해도 user가 없는것이다.

 

listener는 별개의 쓰레드로 동작하고, 트랜잭션도 별개이기때문에, user가 커밋되지않은 시점이기에, user는 null로 조회된다.

 

이것을 해결하는 방법은 크게는 2가지정도가 있다. (더있나, 좀더 생각을 해봐야지)

한가지는 트랜잭션아웃박스 패턴처럼, user테이블이 실제 커밋된 시점까지 폴링을 하는 방법이다.

이건 사실 복잡하고, 폴링에 따른 리소스의 낭비가 심하기에 좋은 방법은 아니라고 생각해서, 사용하지 않는것이 낫다고 생각이 든다.

(패쓰!!!!)

(사실 복잡한건..커서 관리가 귀찮고, 분산처리도 해줘야한다. 쿼리에 skip locked 같은것 옵션줘서,

락프리하게도 만들어야하지만, 락프리기능을 디비에 의존하는 방법이라 좀처럼 맘에 들지는 않는다...)

 

나머지 한가지 방법은 TransactionalEventListener를 이용하는법이다.

카프카 이벤트를 바로 쏘지않고, ApplicationEventPublisher를 이용해 이벤트를 발생시키고,

TransactionalEventListener 함수에서 한번 받아서 카프카 이벤트를 쏘는 방식이다.

 

이때 TransactionalEventListener 함수에서 TransactionPhase.AFTER_COMMIT 옵션을 주어,

커밋이 완료된 시점에 이벤트가 발생하고, 리스닝할수있게 설정하면,

실제로 TransactionalEventListener 함수로 이벤트를 받았을때는 이미 saveUser는 커밋이 완료된 상태를 보장하기에,

이후에 카프카로 이벤트를 발행하고, 리스닝하는 listener함수에서는 user를 조회하면, 값을 조회할수있게 된다.

 

이렇게 커밋이후 시점에 이벤트를 발행하도록 TransactionalEventListener을 이용하는 방법을 사용하는게,

그나마 깔끔한 방법이라고 생각된다.

(참고로 @Async처리도 비슷한 방법을 사용하면, 트랜잭션 이슈를 해결할수있다 :)

 

트랜잭션과 비동기 처리는 참, 생각할것이 많고, 잠깐만 방심하면, 구멍을 만들어낸다.

구멍이 만들어진 상태에서는 디버깅이 가장어려운것이 비동기,트랜잭션 처리이기에,

프로젝트/개발 초기에 이런 부분은 유심히 잘챙겨야 한다.