From 14883ebd84346d6b5f15ecc6e1fae0e11f2836fa Mon Sep 17 00:00:00 2001
From: ahnsugyeong <ahnsugyeong@gmail.com>
Date: Mon, 28 Oct 2024 19:55:54 +0900
Subject: [PATCH] [Fix] Add concurrency setting

---
 .../waggle/global/config/kafka/KafkaConsumerConfig.java        | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/src/main/java/com/example/waggle/global/config/kafka/KafkaConsumerConfig.java b/src/main/java/com/example/waggle/global/config/kafka/KafkaConsumerConfig.java
index 4bcd3eac..4873c597 100644
--- a/src/main/java/com/example/waggle/global/config/kafka/KafkaConsumerConfig.java
+++ b/src/main/java/com/example/waggle/global/config/kafka/KafkaConsumerConfig.java
@@ -12,6 +12,7 @@
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.support.serializer.JsonDeserializer;
 
 import java.util.Map;
@@ -27,6 +28,8 @@ public class KafkaConsumerConfig {
     ConcurrentKafkaListenerContainerFactory<String, ChatMessageDto> kafkaListenerContainerFactory() {
         ConcurrentKafkaListenerContainerFactory<String, ChatMessageDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
         factory.setConsumerFactory(consumerFactory());
+        factory.setConcurrency(6);  // 파티션 개수에 맞춰 스레드 설정
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);  // 수동 커밋 모드
         return factory;
     }