[Kafka] producer 동작 원리(kafka advertised_listener, kafka dns lookup)

 

현업에서 Kafka client와 broker를 연결하기 위해 알아야 할 점들에 대해 살펴본다.

설명을 돕기 위해 kafka go client 중 Shopify/sarama 라이브러리를 사용하였다.

 

1. Kafka cluster에 대한 metadata 요청.

 

brokers := []string{"1.2.3.4:9092", "1.2.3.5:9092"}

config := sarama.NewConfig()

producer, _ := sarama.NewAsyncProducer(brokers, config)

 

보통 위와 같이 broker 및 config에 대한 정보를 설정하고 NewProducer를 통해 metadata를 가져온다.

kafka client는 전달받은 brokers slice를 순차적으로 탐색하며 metadata를 가지고 오기 위한 request를 발생시키고, kafka cluster를 구성하고있는 모든 broker에는 해당 metadata를 response 해줄 수 있는 기능이 있다.

따라서 위 코드의 경우 먼저 1.2.3.4 broker에 metadata를 요청하고, 1.2.3.4 broker가 정상 상태여서 metadata를 response해주면 나머지 뒤에 있는 1.2.3.5에는 요청을 보내지 않고 바로 해당 metadata를 기반으로 producer 객체를 만든다. 

    작성한 ip를 가진 broker 두 개가 모두 down상태라면,  kafka client는 metadata를 가지고 올 수 없게되고 application은 에러를 리턴한다. kafka cluster를 구성하는 broker가 두 개면 어쩔 수 없지만, 만약 kafka cluster를 구성하는 broker가 20개이고, 그 중에 두 개만 client에 ip로 적어두었다면 kafka의 HA 기능을 제대로 살리지 못한 채 사용하고 있는 셈이 된다. 따라서 만약 kafka의 HA 기능을 100%로 활용하고 싶다면, broker를 구성하고 있는 모든 ip를 적어주는 것이 좋다.

    그런데 만약 kafka를 운영하는 곳에서 1.2.3.4 ip를 사용하던 서버를 내리고, 10.20.30.40 ip를 사용하는 새로운 서버를 생성하여 kafka cluster에 추가하면 어떻게 될까?

client에서는 그 때마다 정보를 공유받아야하고, broker 부분에 적힌 ip를 수정한 뒤 application을 새롭게 빌드, 배포를 해야 한다. application과 kafka 사이에 dependency가 생기는 것이다.

이를 해결하기 위해서는 multi ip DNS를 사용해야 한다.

카프카를 관리하는 부서는 카프카에 대한 DNS까지 관리를하고, 해당 도메인에 카프카 클러스터를 구축하는 ip를 등록해 둔다.

    DNS를 가지고와서 사용하는 부분은 구현부마다 다르기는한데, go의 conn을 확인해본 결과 go는 DNS에 multi ip가 붙어있는 경우 모든 row를 가져오고, 순차적으로 request를 날려 가능한 connection인지 확인을 한다. 

이런 식으로 구축할 경우 Kafka를 운영하는 부서와 kafka를 사용하는 부서의 dependency가 제거될 수 있다.

(DNS 캐싱에 대한 문제가 있을 수 있는데, 알아볼 것.)

 

 2. metadata broker addrs 문제

kafka를 전문적으로 운영하는 팀이 없거나 직접 구축해야 하는 경우, metadata broker addrs에 broker의 hostname이 들어있는 경우가 있다. producer는 메세지를 보내야 할 때 이 metadata broker addrs에 저장된 string 값을 사용하게 되는데, 이 때 hostname을 사용하게 되면 당연히 dns resolve가 에러를 뱉게된다. 이를 해결하기 위해서는 producer 서버의 hosts파일에 해당 hostname에 대한 ip를 적어두거나, broker의 세팅을 바꿔야 한다.

 kafka cluster에서 해당 broker addrs에 broker의 hostname을 주는 이유는 server.properties에 advertised_listener 설정을 해주지 않아서이다. 이 부분을 설정해주지 않으면 default로 hostname을 리턴해주게 되는데, 여기를 공인 ip로 변경하면 된다. (사설 ip로 설정하면 당연히 또 안됨)