feat : 장애 대응 및 재처리 로직을 적용한 고가용성 알림 시스템 구축

.gitignore

@@ -43,6 +43,7 @@ src/main/resources/application.properties
 docker-compose.dev.yml
 /data/
 /db/
+/replicaset
 
 # 환경변수
 .env

build.gradle

@@ -55,7 +55,7 @@ dependencies {
 
     // DB
     runtimeOnly 'com.h2database:h2'
-    runtimeOnly 'com.mysql:mysql-connector-j'
+    implementation 'com.mysql:mysql-connector-j'
 
     annotationProcessor 'org.projectlombok:lombok'
 
@@ -120,6 +120,10 @@ dependencies {
 
     //AES 암호화
     implementation 'javax.xml.bind:jaxb-api:2.3.1'
+
+    // Circuit Breaker
+    implementation 'io.github.resilience4j:resilience4j-spring-boot3:2.2.0'
+    implementation 'io.github.resilience4j:resilience4j-circuitbreaker:2.2.0'
 }
 
 tasks.named('test') {

restart-mongodb-replica.sh

@@ -0,0 +1,53 @@
+DATA_FILE_PATH="./replicaset"
+DOCKER_FILE_PATH="./docker-compose.dev.yml"
+MONGO_PRIMARY_NAME="rs01p"
+REPLICA_INIT_FILE_PATH="./scripts/rs-init.sh"
+MONGO_CREATE_USER_FILE_PATH="./scripts/mongo-create-user.sh"
+
+UP_CONTAINER_DELAY=10
+REPLICA_CONFIG_DELAY=25
+
+echo "****************** Reset docker container Shell Script ******************"
+echo "Data File Path: ${DATA_FILE_PATH}"
+echo "Docker File Path: ${DOCKER_FILE_PATH}"
+echo "MongoDB Primary name: ${MONGO_PRIMARY_NAME}"
+echo "Replica set init Script File Path: ${REPLICA_INIT_FILE_PATH}"
+echo "Mongo create user file path: ${MONGO_CREATE_USER_FILE_PATH}"
+
+sleep 1;
+
+echo "****************** Stop docker container ******************"
+docker-compose -f ${DOCKER_FILE_PATH} stop
+echo "****************** Completed Stop docker container ******************"
+
+sleep 1;
+
+echo "****************** Down docker container ******************"
+docker-compose -f ${DOCKER_FILE_PATH} down
+echo "****************** Completed Down docker container ******************"
+
+sleep 1;
+
+echo "****************** Remove Data ******************"
+rm -rf ${DATA_FILE_PATH}
+echo "****************** Completed Remove Data ******************"
+
+sleep 1;
+
+echo "****************** Up docker container ******************"
+docker-compose -f ${DOCKER_FILE_PATH} up -d 
+echo "****************** Completed Up docker container ******************"
+
+echo "****** Waiting for ${UP_CONTAINER_DELAY} seconds ******"
+sleep $UP_CONTAINER_DELAY;
+
+echo "****************** Run Replica Set Shell Script ******************"
+docker exec -i ${MONGO_PRIMARY_NAME} bash < ${REPLICA_INIT_FILE_PATH}
+
+echo "****** Waiting for ${REPLICA_CONFIG_DELAY} seconds for replicaset configuration to be applied ******"
+sleep $REPLICA_CONFIG_DELAY
+
+echo "****************** Run Create DB User Shell Script ******************"
+docker exec -i ${MONGO_PRIMARY_NAME} bash < "${MONGO_CREATE_USER_FILE_PATH}"
+
+echo "****************** Completed Replica Shell Script ******************"

scripts/mongo-create-user.sh

@@ -0,0 +1,14 @@
+mongosh --port 10021 <<EOF
+use admin;
+db.createUser({
+    user: "mongo",
+    pwd: "mongo123",
+    roles: [
+      {
+        role: "dbOwner",
+        db: "ezcode",
+      },
+    ],
+});
+db.getUsers();
+EOF

scripts/rs-init.sh

@@ -0,0 +1,28 @@
+mongosh --port 10021 <<EOF
+use ezcode;
+var config = {
+    "_id": "rs01",
+    "version": 1,
+    "members": [
+        {
+            "_id": 1,
+            "host": "rs01p:10021",
+            "priority": 2
+        },
+        {
+            "_id": 2,
+            "host": "rs01s:10022",
+            "priority": 1
+        },
+        {
+            "_id": 3,
+            "host": "rs01a:10023",
+            "priority": 0
+        }
+    ]
+};
+rs.initiate(config);
+rs.status();
+db.setProfilingLevel(2);
+db.getProfilingStatus();
+EOF

src/main/java/org/ezcode/codetest/application/notification/exception/NotificationException.java

@@ -24,4 +24,13 @@ public NotificationException(ResponseCode responseCode, String message) {
 		this.httpStatus = responseCode.getStatus();
 		this.message = responseCode.getMessage() + " : " + message;
 	}
+
+	public NotificationException(ResponseCode responseCode, Throwable cause, String message) {
+		super();
+
+		this.responseCode = responseCode;
+		this.httpStatus = responseCode.getStatus();
+		super.initCause(cause);
+		this.message = message;
+	}
 }

src/main/java/org/ezcode/codetest/application/notification/exception/NotificationExceptionCode.java

@@ -12,7 +12,8 @@ public enum NotificationExceptionCode implements ResponseCode {
 
 	NOTIFICATION_CANNOT_FIND_EVENT_TYPE(false, HttpStatus.INTERNAL_SERVER_ERROR, "해당 이벤트 타입의 mapper를 찾을 수 없습니다."),
 	NOTIFICATION_CONVERT_MESSAGE_ERROR(false, HttpStatus.INTERNAL_SERVER_ERROR, "메시지 변환 과정에서 에러가 발생했습니다."),
-	NOTIFICATION_NOT_FOUND(false, HttpStatus.NOT_FOUND, "해당 ID의 notification 데이터를 찾지 못했습니다")
+	NOTIFICATION_NOT_FOUND(false, HttpStatus.NOT_FOUND, "해당 ID의 notification 데이터를 찾지 못했습니다"),
+	NOTIFICATION_DB_ERROR(false, HttpStatus.INTERNAL_SERVER_ERROR, "알림 데이터 저장 중 문제 발생. 서킷 브레이커가 열렸습니다.")
 	;
 
 	private final boolean success;

src/main/java/org/ezcode/codetest/infrastructure/mongo/config/MongoTransactionConfig.java

@@ -0,0 +1,15 @@
+package org.ezcode.codetest.infrastructure.mongo.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.mongodb.MongoDatabaseFactory;
+import org.springframework.data.mongodb.MongoTransactionManager;
+
+@Configuration
+public class MongoTransactionConfig {
+
+	@Bean
+	public MongoTransactionManager transactionManager(MongoDatabaseFactory mongoDatabaseFactory) {
+		return new MongoTransactionManager(mongoDatabaseFactory);
+	}
+}

src/main/java/org/ezcode/codetest/infrastructure/notification/event/NotificationSavedEvent.java

@@ -0,0 +1,12 @@
+package org.ezcode.codetest.infrastructure.notification.event;
+
+import org.ezcode.codetest.infrastructure.notification.dto.NotificationResponse;
+
+public record NotificationSavedEvent(
+
+	String principalName,
+
+	NotificationResponse response
+
+) {
+}

src/main/java/org/ezcode/codetest/infrastructure/notification/model/NotificationProcessLog.java

@@ -0,0 +1,72 @@
+package org.ezcode.codetest.infrastructure.notification.model;
+
+import java.time.LocalDateTime;
+
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import org.springframework.data.annotation.Id;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+@Document(collection = "notification_process_log")
+@Getter
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@AllArgsConstructor
+public class NotificationProcessLog {
+
+	@Id
+	private String messageId;
+
+	private String payload;
+
+	private ProcessStatus status;
+
+	private int retryCount;
+
+	private String errorMessage;
+
+	private LocalDateTime lastAttemptAt;
+
+	private LocalDateTime createdAt;
+
+	public enum ProcessStatus {
+		PENDING, SUCCESS, FAILED, PERMANENTLY_FAILED
+	}
+
+	public static NotificationProcessLog of(String messageId, String payload) {
+
+		return new NotificationProcessLog(
+			messageId,
+			payload,
+			ProcessStatus.PENDING,
+			0,
+			null,
+			LocalDateTime.now(),
+			LocalDateTime.now()
+		);
+	}
+
+	public void markAsSuccess() {
+		this.status = ProcessStatus.SUCCESS;
+		this.lastAttemptAt = LocalDateTime.now();
+		this.errorMessage = null;
+	}
+
+	public void markAsFailed(String errorMessage, int maxRetries) {
+		this.retryCount++;
+		this.lastAttemptAt = LocalDateTime.now();
+		this.errorMessage = errorMessage;
+
+		if (this.retryCount >= maxRetries) {
+			this.status = ProcessStatus.PERMANENTLY_FAILED;
+		} else {
+			this.status = ProcessStatus.FAILED;
+		}
+	}
+
+	public void updateLastAttempt() {
+		this.lastAttemptAt = LocalDateTime.now();
+	}
+}

src/main/java/org/ezcode/codetest/infrastructure/notification/model/NotificationQueueConstants.java

@@ -4,6 +4,8 @@ public final class NotificationQueueConstants {
 
 	private NotificationQueueConstants() {}
 
+	public static final String CUSTOM_HEADER_MESSAGE_ID = "custom_message_id";
+
 	public static final String NOTIFICATION_QUEUE_CREATE = "notification.queue.create";
 	public static final String NOTIFICATION_QUEUE_LIST = "notification.queue.list";
 	public static final String NOTIFICATION_QUEUE_MARK_READ = "notification.queue.read";

src/main/java/org/ezcode/codetest/infrastructure/notification/publisher/NotificationEventPublisher.java

@@ -2,6 +2,8 @@
 
 import static org.ezcode.codetest.infrastructure.notification.model.NotificationQueueConstants.*;
 
+import java.util.UUID;
+
 import org.ezcode.codetest.application.notification.event.*;
 import org.ezcode.codetest.application.notification.exception.NotificationException;
 import org.ezcode.codetest.application.notification.exception.NotificationExceptionCode;
@@ -27,29 +29,31 @@ public class NotificationEventPublisher implements NotificationEventService {
 	public void notify(NotificationCreateEvent dto) {
 
 		sendMessage(NOTIFICATION_QUEUE_CREATE, dto);
-		// publisher.publishEvent(dto);
 	}
 
 	@Override
 	public void notifyList(NotificationListRequestEvent dto) {
 
 		sendMessage(NOTIFICATION_QUEUE_LIST, dto);
-		// publisher.publishEvent(dto);
 	}
 
 	@Override
 	public void setRead(NotificationMarkReadEvent dto) {
 
 		sendMessage(NOTIFICATION_QUEUE_MARK_READ, dto);
-		// publisher.publishEvent(dto);
 	}
 
 	private void sendMessage(String destination, Object data) {
 
 		try {
 			String jsonMessage = objectMapper.writeValueAsString(data);
 
-			jmsTemplate.convertAndSend(destination, jsonMessage);
+			String customMessageId = "ID-" + UUID.randomUUID();
+			jmsTemplate.convertAndSend(destination, jsonMessage, message -> {
+				message.setStringProperty(CUSTOM_HEADER_MESSAGE_ID, customMessageId);
+				return message;
+			});
+
 			log.info("알림 메시지 전송 성공 ({}) : {}", destination, jsonMessage);
 		} catch (JsonProcessingException ex) {
 			log.error("알림 메시지 변환 및 전송 실패 : {}", ex.getMessage());

src/main/java/org/ezcode/codetest/infrastructure/notification/repository/NotificationProcessLogRepository.java

@@ -0,0 +1,12 @@
+package org.ezcode.codetest.infrastructure.notification.repository;
+
+import java.util.List;
+
+import org.ezcode.codetest.infrastructure.notification.model.NotificationProcessLog;
+import org.springframework.data.mongodb.repository.MongoRepository;
+
+public interface NotificationProcessLogRepository extends MongoRepository<NotificationProcessLog, String> {
+
+	// 재시도할 작업들을 찾는 쿼리 메서드
+	List<NotificationProcessLog> findByStatusAndRetryCountLessThan(NotificationProcessLog.ProcessStatus status, int maxRetries);
+}