kafka 사용시 한 쪽 파티션에만 메시지가 도착하는 경우

kafka를 통해 공부를 하던 중 메시지를 발행하면 계속 한쪽으로만 쌓이는 상황이 발생하였습니다.

kafka의 produce시 작동원리
1. key가 있는 경우 키를 hash한 값을 통해 파티션 분배
2. key가 없다면 round Robbing으로 적재 -> sticky 방식으로 적재

기본 코드

@Slf4j
public class SimpleProducer {
	private final static String TOPIC_NAME = "test";
	private final static String BOOTSTRAP_SERVERS = "localhost:9092";

	@SneakyThrows
	public static void main(String[] args) {
		Properties configs = new Properties();

		configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
		configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

		KafkaProducer<String, String> producer = new KafkaProducer<>(configs);

		String messageValue = "testMessage";

		for (int i = 0; i < 30; i++) {
			ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, messageValue + i);
			producer.send(record);
		}

		producer.flush();
		producer.close();
	}
}

위 코드 실행 전 kafka를 실행하여 kafka-topics.sh --create --bootstrap-server kafka1:29092 --topic test --partitions 3 명령어를 사용하여 3개의 파티션을 가지는 topic을 만들었습니다.

단순히 해당 topic에 메시지를 30개를 보내는 코드입니다.

해당 데이터가 어떻게 저장이 되었는지 확인하기 위해 kafka-ui를 이용해서 확인해보면 아래와 같이 한 쪽 파티션에만 저장이 된 것을 알 수 있습니다.

기본 코드 실행 결과

위 코드의 결과가 나온 이유로는 partitioner.class를 설정하지 않은 경우 기본 동작대로 움직이게 되는데 https://kafka.apache.org/documentation/#producerapi 를 통해 partitioner.class를 확인해 보면 기본 설정이 없는 경우에는 batch.size만큼 partition에 저장이 된다고 적혀있습니다.

정말로 그런지 batch.size를 설정해주고 나서 실행을 해보면

configs.put(ProducerConfig.BATCH_SIZE_CONFIG, 1); // <- 추가

batch size 설정

실제로 데이터들이 나뉘어서 들어간 것을 확인 할 수 있습니다.

partion class 변경

partition 동작원리는 아래와 같은 방식으로 동작하도록 되어 있습니다.
1. 파티션이 지정되지 않았지만 키가 존재하는 경우, 키의 해시를 기반으로 파티션을 선택합니다.
2. 파티션과 키가 모두 지정되지 않은 경우, 적어도 batch.size 바이트가 파티션에 생성될 때까지 변경되지 않는 '스티키 파티션'을 선택합니다.

만약 기본 동작대로 하지 않고 RoundRobinPartitioner를 사용하고 싶으시다면 아래 코드를 넣어주시면 됩니다.

configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, 
RoundRobinPartitioner.class.getName()); // <- 추가

이 밖에도 자신만의 파티셔닝 전략을 적용하고 싶으시다면 Partitioner 클래스를 수정해서 위 class를 대체해서 넣으시면 됩니다.

추가

위 방식을 해도 생각과 다르게 진행된다면 batch size 옵션을 줄여보시면 될 것 같습니다. (카프카 공식 홈페이지에 가시면 여러 옵션이 있는데 이를 활용해 보시면 될 것 같습니다.)

참고자료

kafka