1. 배경
Java에서 카프카 프로듀서를 개발하며 디버깅을 하다 이상한 현상을 겪게 되어 기록의 목적으로 이 글을 작성하게되었습니다.
package com.example.kafka;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
public class SimpleProducer {
public static void main(String[] args) {
String topicName = "simple-topic";
// KafkaProducer configuration setting
Properties props = new Properties();
//bootstrap.servers, key.serializer.class, value.serializer.class
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// KafkaProducer 객체 생성
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
// KafkaProducer message send
kafkaProducer.send(new ProducerRecord<>(topicName, "hello 1"));
kafkaProducer.send(new ProducerRecord<>(topicName, "hello world")
kafkaProducer.flush();
kafkaProducer.close();
}
}
우선 KafkaProducer는 내부적으로 전달 받은 메시지를 serializing 후 Buffer에 담아둔 후 이것을 한 번에 보내도록 설계되어있습니다.
따라서 KafkaProducer.flush()에 BreakPoint를 두고 Buffer에 해당 값들이 잘 담겨있는지 확인했습니다.
그 결과 KafkaProducer.accumulator의 topicInfoMap에 value가 recordCount=2를 보여주고 있다는 것을 눈으로 확인했습니다.
여기서 flush()까지 실행 시 당연히 정삭적으로 consumer에서 메시지를 읽을 수 있습니다.
물론 Kafka는 꼭 flush() 시점에만 메시지를 발송하는 것은 아닙니다. Buffer에 담아둔 시간이 특정 시간을 넘기거나, 버퍼가 가득찬 경우는 직접 flush()를 하지 않아도 메시지를 자동으로 발송하게 됩니다.
이를 테스트하기 위해 pros에 설정 하나를 추가했습니다.
props.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1");
에지 KafkaProducer.send()로 메시지를 전달하고, 내부적으로 이 메시지가 buffer에 담기게 된다면, linger.ms가 아주 작은 값(1ms)이기에 바로 전송이 될 것 입니다.
kafkaProducer.send(new ProducerRecord<>(topicName, "hello 1"));
kafkaProducer.send(new ProducerRecord<>(topicName, "hello world"));
kafkaProducer.flush();이것을 확인하기 위해 위 세 줄에 BreakPoint를 설정 후 KafkaProducer.flush() 위치에서 Buffer가 비었는지 확인해보았습니다.
2. 문제
그 결과 기대와 다르게 여전히 recordCount=2로 Buffer가 비워지지 않은 것을 확인했습니다. 분명 버퍼 대기 시간을 굉장히 짧은 시간(1ms)으로 두었기에 바로 전송이되어야 하는데 전송되지 않는 것이 이상하여 자세히 알아보았습니다.
먼저 KafkaProducer는 buffer에 보낼 메시지를 저장하고나면 ioThread가 해당 버퍼에서 메시지를 꺼내 Broker로 전달합니다.
이 과정에서 메인 스레드와 IO Thread가 동시에 같은 buffer를 바라보고 있어 나타나는 문제였습니다.
2-1. 첫 번째 범인: Lock를 쥐고 멈춘 메인 스레드
카프카의 send() 로직 내부에는 멀티스레드 환경에서 안전하게 데이터를 쌓기 위해 synchronized 블록이 구현되어 있습니다.
- 현상: 메인 스레드가 send() 내부에서 디버깅 중(Step Into)이라면, 이 Lock을 쥔 채로 멈춰 있는 상태가 됩니다.
- 결과: 백그라운드에서 데이터를 실어 나르는 ioThread는 CPU 자원이 충분해도, 이 열쇠가 없어 공유 메모리(Accumulator)에 접근하지 못하고 문밖에서 대기하게 됩니다.
2-2. 두 번째 범인: 디버거의 '모든 스레드 중지(Suspend All)' 설정
"아니, 나는 send()를 빠져나와 flush() 줄에서 대기 중인데 왜 여전히 안 가죠? 락은 반납했을 텐데!"라는 의문이 생깁니다. 이때는 디버거의 기본 설정을 의심해야 합니다.
- Suspend All: 인텔리제이(IntelliJ)의 기본 브레이크 포인트 설정은 특정 스레드만 멈추는 게 아니라, 프로그램 전체 스레드의 시간을 멈춰버립니다.
- 상황: 메인 스레드가 flush() 줄에 걸려 멈춰 있다면, 데이터를 쏘아야 할 ioThread까지 디버거가 같이 얼려버린 것입니다. 락은 풀렸지만, 정작 일을 해야 할 일꾼이 멈춰 있으니 메시지는 전송되지 않습니다.
4. 해결책: 디버깅 중에도 전송되는 것을 보고 싶다면?
가장 확실한 방법은 브레이크 포인트의 속성을 바꾸는 것입니다.
- 브레이크 포인트(빨간 점) 우클릭.
- Suspend 옵션을 All에서 Thread로 변경.
- 이렇게 하면 메인 스레드만 멈추고 ioThread는 자유롭게 돌아다니며 데이터를 브로커로 보낼 수 있게 됩니다.