test(be): 파티션 개수 증가 및 ack 설정 변경

Author: ki-met-hoon

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/service/KafkaProducer.java

@@ -24,9 +24,11 @@ public class KafkaProducer {
 
     public void sendChatMessage(ChatMessageToKafka chatMessageToKafka, Long channelId) {
         try {
+            String randomKey = String.valueOf(ThreadLocalRandom.current().nextInt(1, 6));
+
             String jsonChatMessage = objectMapper.writeValueAsString(chatMessageToKafka);
 
-            kafkaTemplate.send(topicChat, String.valueOf(channelId), jsonChatMessage)
+            kafkaTemplate.send(topicChat, randomKey, jsonChatMessage)
                     .whenComplete((result, ex) -> { //키 값 설정으로 순서 보장, 실시간성이 떨어짐, 고민해봐야 할 부분
                         if (ex == null) {
                             log.info("Kafka message sent: {}", result.toString());

src/backend/state_server/src/main/java/com/jootalkpia/state_server/service/KafkaConsumer.java

@@ -21,7 +21,8 @@ public class KafkaConsumer {
 
     @KafkaListener(
             topics = "${topic.chat}",
-            groupId = "${group.status}"
+            groupId = "${group.status}",
+            concurrency = "5"
     )
     public void processState(String kafkaMessage) {
         ObjectMapper mapper = new ObjectMapper();

src/backend/state_server/src/main/java/com/jootalkpia/state_server/service/StateService.java

@@ -32,11 +32,7 @@ public PushMessageToKafka createPushMessage(JsonNode commonNode, JsonNode messag
     }
 
     public Set<String> findNotificationTargets(String channelId, String userId) {
-        Set<String> testSessions = stringOperRedisTemplate.opsForSet().members("user:sessions");
-
-        log.info(testSessions.toString());
-
-        return testSessions;
+        return stringOperRedisTemplate.opsForSet().members("user:sessions");
 
 //        Set<String> subscriber = findSubscribers(channelId);
 //        Set<String> onlineSessions = findOnlineSessions(userId, subscriber);

src/backend/state_server/src/main/resources/application.yml

@@ -5,9 +5,14 @@ spring:
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
     producer:
-      acks: all # 멱등성 프로듀서를 위해 all로 설정, 0 또는 1이면 enable.idempotence=true 불가
+      acks: 0
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.apache.kafka.common.serialization.StringSerializer
+      batch-size: 65536  # 64KB
+      properties:
+        linger.ms: 20    # 20ms
+        buffer.memory: 67108864  # 64MB로 버퍼 메모리도 증가
+        compression.type: snappy
   data:
     redis:
       host: ${REDIS_HOST}