Kafka Producer의 config(linger, acks, compression)

Kafka를 통해 데이터를 송신해야하는 케이스가 생겼다.

비지니스 로직상 빈도는 낮으나(한달에 한번도 호출 안한다.) 한번 발생하면 한번에 약 3만건의 데이터를 전송해야한다.

 

개발 후 consumer측과 테스트를 진행하던 도중 문제가 생겼는데

데이터를 단건 씩 보낼때는 이상이 없었으나 3만건을 반복문으로 계속 보냈더니 수신한 데이터의 건은 1,2000건이었다.....

 

LAG나 다른 kafak의 상태를  보고싶었지만 kafkacat과 같은 툴을 반입할 수 없고, kafka 서버에 직접 터미널로 접속할 수 없는 상태였다.

 

일단 송신과정에서 누락이 된것인가 싶어서 kafka producer의 ACKS옵션을 변경하였다.

 

acks

application.properties

spring.kafka.producer.acks=-1

 

acks의 옵션은 총 3가지로 -1, 0, 1이다.

• 0 : 프로듀서가 kafka에 메시지를 전송하고 kafka로부터 잘 전송되었는지 응답을 받지 않는다.
• 1 :  프로듀서가 kafka에 메시지를 전송하고 kafka의 leader가 잘 받았는지 응답을 받는다.
• -1(all) : 프로듀서가 kafka에 메시지를 전송하고 kafka의 leader, follower가 잘 받았는지 응답을 받는다.

옵션 1, -1은 응답으로 실패를 받는다면 spring kafka가 재전송을 한다.

 

---

akcs를 적용후에도 동일한 증상이 발생.

단건 전송시에는 문제가 없었으니 데이터를 100건마다 Thread.sleep을 적용해보았다.

 

if (count % 100 == 0){
	Thread.sleep(1000)
}

 

수신측에서 모든 데이터가 들어왔다고 한다.

하지만 kafka 전송하는 로직이 @Service로 싱글톤 내부에 있는 메소드다.

 

싱글톤 내부의 메소드에서 Thread.sleep을 사용하게 되면 해당 로직이 끝나기 전까지는 블록상태가 된다.(맞나...?)

 

---

일단 kafka로 send하는 로직에 시간을 늘리면 해결된다는 부분은 확인을 했으니 방법을 찾아봤다.

 

Excuters로 별도의 스레드에 할당 할까 하였지만?

Kafka Producer의 다른 옵션들이 있는것을 확인

 

---

batch.size

메시지보관 사이즈로 linger의 옵션 시간동안 메시지를 쌓는다.

쌓인 메시지가 size보다 작으면 linger의 옵션까지 대기했다가 전송한다.

 

디폴트:16384

 

linger

linger.ms는 데이터 전송까지의 시간을 의미한다.

기본값은 0으로 데이터를 즉시 전송한다.

 

linger.ms = 10라면 10ms 마다 send한다.

 

단 batch.size만큼 데이터가 쌓여있다면 linger옵션을 무시하고 전송한다.

즉 batch.size가 꽉 찰 때 까지 기다리는 옵션

 

spring.kafka.producer.properties.linger.ms=10
// 혹은
spring.kafka.producer.linger.ms=10

Spring Kafka의 버전마다 해당 옵션 키값이 다른 듯 하다.

 

compression

메시지를 압축하는 옵션이다.

 

메시지를 압축하면 하지 않는 것 보다 발송 지연이 생기지만 비지니스 로직상 허용 가능한 부분이기에 추가하였다.

압축을 통하여 batch.size에 쌓이는 메세지의 수가 늘어나도록 설정하였다.

 

spring.kafka.producer.compression-type=snappy

 

---