hotfix : 제출 제대로 안 되는 부분 해결 (제출 씹힘 현상) (#68)

* refactor : 쓰레드 풀 추가

* refactor : 예외처리 고도화

* refactor : Redis Stream Cleanup

* feat : Discord 알림

* fix : 비동기 메서드에서 시큐리티 컨텍스트 전파 되지 않는 현상 수정

* fix : CI 환경 변수 추가

* refactor : 로깅 처리 변경

* refactor : 예외 처리 변경

* refactor : 예외처리 고도화

# Conflicts:
#	src/main/java/org/ezcode/codetest/common/base/exception/GlobalExceptionHandler.java

* fix : 비동기 메서드에서 시큐리티 컨텍스트 전파 되지 않는 현상 수정

* refactor : 토끼 리뷰 반영

* refactor : 토끼 리뷰 반영

* hotfix : 제출 안 되는 부분 해결

Author: thezz9

src/main/java/org/ezcode/codetest/application/submission/service/SubmissionService.java

@@ -77,7 +77,8 @@ public SseEmitter enqueueCodeSubmission(Long problemId, CodeSubmitRequest reques
 			emitterStore.remove(emitterKey);
 		});
 
-        emitterStore.save(emitterKey, emitter);
+		log.info("[SSE 저장] emitterKey: {}", emitterKey);
+		emitterStore.save(emitterKey, emitter);
 
         queueProducer.enqueue(
 			new SubmissionMessage(emitterKey, problemId, request.languageId(), authUser.getId(), request.sourceCode())
@@ -90,6 +91,7 @@ public SseEmitter enqueueCodeSubmission(Long problemId, CodeSubmitRequest reques
     public void submitCodeStream(SubmissionMessage msg) {
 		try {
             log.info("[Submission RUN] Thread = {}", Thread.currentThread().getName());
+			log.info("[큐 수신] SubmissionMessage.emitterKey: {}", msg.emitterKey());
 			User user = userDomainService.getUserById(msg.userId());
             Language language = languageDomainService.getLanguage(msg.languageId());
             ProblemInfo problemInfo = problemDomainService.getProblemInfo(msg.problemId());
@@ -126,7 +128,9 @@ public void submitCodeStream(SubmissionMessage msg) {
 			}
 
 			if (!context.latch().await(60, TimeUnit.SECONDS)) {
-				throw new SubmissionException(SubmissionExceptionCode.TESTCASE_TIMEOUT);
+				emitter.completeWithError(new SubmissionException(SubmissionExceptionCode.TESTCASE_TIMEOUT));
+				emitterStore.remove(msg.emitterKey());
+				return;
 			}
 
 			emitter.send(SseEmitter.event()

src/main/java/org/ezcode/codetest/infrastructure/event/config/RedisStreamConfig.java

@@ -7,6 +7,7 @@
 import org.ezcode.codetest.infrastructure.event.listener.RedisJudgeQueueConsumer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.stream.MapRecord;
 import org.springframework.data.redis.connection.stream.ReadOffset;
@@ -43,6 +44,20 @@ public void initConsumerGroup() {
 				));
 			}
 
+			try {
+				log.info("Redis Consumer Group의 기존 컨슈머 삭제 시도");
+				redisTemplate.execute((RedisConnection connection) -> {
+					connection.xGroupDelConsumer(
+						"judge-queue".getBytes(),
+						"judge-group",
+						"consumer-1"
+					);
+					return null;
+				});
+			} catch (Exception e) {
+				log.warn("DELCONSUMER 중 오류 발생: {}", e.getMessage());
+			}
+
 			redisTemplate.opsForStream().createGroup("judge-queue", ReadOffset.latest(), "judge-group");
 
 			log.info("Redis Stream 'judge-queue'에 대한 Consumer Group 'judge-group'을 생성했습니다.");
@@ -57,7 +72,7 @@ public void initConsumerGroup() {
 		}
 	}
 
-	@Bean
+	@Bean(destroyMethod = "stop")
 	public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(
 		RedisConnectionFactory factory,
 		RedisJudgeQueueConsumer consumer
@@ -68,7 +83,7 @@ public StreamMessageListenerContainer<String, MapRecord<String, String, String>>
 				.StreamMessageListenerContainerOptions
 				.builder()
 				.executor(consumerExecutor)
-				.pollTimeout(Duration.ofSeconds(1))
+				.pollTimeout(Duration.ofSeconds(2))
 				.build();
 
 		StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =

src/main/java/org/ezcode/codetest/infrastructure/event/listener/RedisJudgeQueueConsumer.java

@@ -35,11 +35,14 @@ public void onMessage(MapRecord<String, String, String> message) {
 		);
 
 		try {
+			log.info("[컨슈머 수신] {}", msg.emitterKey());
 			submissionService.submitCodeStream(msg);
-			redisTemplate.opsForStream().acknowledge("judge-group", message);
 		} catch (Exception e) {
 			log.error("채점 메시지 처리 실패: {}", message.getId(), e);
 			throw new SubmissionException(SubmissionExceptionCode.REDIS_SERVER_ERROR);
+		} finally {
+			log.info("[컨슈머 ACK] messageId={}", message.getId());
+			redisTemplate.opsForStream().acknowledge("judge-group", message);
 		}
 	}
 }

src/main/java/org/ezcode/codetest/infrastructure/event/service/RedisJudgeQueueProducer.java

@@ -8,14 +8,17 @@
 import org.springframework.stereotype.Component;
 
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 @Component
 @RequiredArgsConstructor
 public class RedisJudgeQueueProducer implements QueueProducer {
 
 	private final RedisTemplate<String, String> redisTemplate;
 
 	public void enqueue(SubmissionMessage submissionMessage) {
+		log.info("[SSE enqueue] emitterKey: {}", submissionMessage.emitterKey());
 		Map<String, String> map = Map.of(
 			"emitterKey", submissionMessage.emitterKey(),
 			"problemId", submissionMessage.problemId().toString(),

src/main/java/org/ezcode/codetest/infrastructure/sse/InMemoryEmitterStore.java

@@ -16,9 +16,6 @@ public class InMemoryEmitterStore implements EmitterStore {
 	@Override
 	public void save(String key, SseEmitter emitter) {
 		emitterMap.put(key, emitter);
-		emitter.onCompletion(() -> emitterMap.remove(key));
-		emitter.onTimeout(() -> emitterMap.remove(key));
-		emitter.onError((e) -> emitterMap.remove(key));
 	}
 
 	@Override