refactor: (#962) race condition 문제를 rabbitMQ 단일 처리 지점 생성으로 해결

Author: ryujaemin08

dms-main/main-core/src/main/kotlin/team/aliens/dms/common/spi/OutboxPort.kt

@@ -2,11 +2,9 @@ package team.aliens.dms.common.spi
 
 import team.aliens.dms.common.dto.OutboxData
 import team.aliens.dms.common.dto.OutboxStatus
-import java.util.UUID
 
 interface OutboxPort {
     fun save(outbox: OutboxData): OutboxData
-    fun deleteById(id: UUID)
     fun findByStatus(status: OutboxStatus): List<OutboxData>
     fun delete(outbox: OutboxData)
 }

dms-main/main-infrastructure/src/main/kotlin/team/aliens/dms/event/handler/DeviceTokenEventHandler.kt

@@ -1,7 +1,6 @@
 package team.aliens.dms.event.handler
 
 import com.fasterxml.jackson.databind.ObjectMapper
-import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 import org.springframework.transaction.event.TransactionPhase
 import org.springframework.transaction.event.TransactionalEventListener
@@ -13,23 +12,18 @@ import team.aliens.dms.contract.remote.rabbitmq.SaveDeviceTokenMessage
 import team.aliens.dms.event.DeleteDeviceTokenEvent
 import team.aliens.dms.event.DeviceTokenEvent
 import team.aliens.dms.event.SaveDeviceTokenEvent
-import team.aliens.dms.thirdparty.messagebroker.NotificationProducer
-import java.util.UUID
 
 @Component
 class DeviceTokenEventHandler(
     private val outboxPort: OutboxPort,
-    private val notificationProducer: NotificationProducer,
     private val objectMapper: ObjectMapper
 ) {
-    private val log = LoggerFactory.getLogger(this::class.java)
-    private val outboxIdsByEventIdentity = ThreadLocal<MutableMap<Int, UUID>>()
 
     @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
     fun saveOutbox(event: DeviceTokenEvent) {
         val (eventType, message) = createMessage(event)
 
-        val saved = outboxPort.save(
+        outboxPort.save(
             OutboxData(
                 id = null,
                 aggregateType = "device_token",
@@ -38,29 +32,6 @@ class DeviceTokenEventHandler(
                 status = OutboxStatus.PENDING
             )
         )
-        val map = outboxIdsByEventIdentity.get() ?: mutableMapOf<Int, UUID>().also {
-            outboxIdsByEventIdentity.set(it)
-        }
-
-        map[System.identityHashCode(event)] = saved.id!!
-    }
-
-    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
-    fun publishMessage(event: DeviceTokenEvent) {
-        val (_, message) = createMessage(event)
-        val map = outboxIdsByEventIdentity.get()
-        val outboxId = map.remove(System.identityHashCode(event))
-
-        try {
-            notificationProducer.sendMessage(message)
-            outboxId?.let { outboxPort.deleteById(it) }
-        } catch (e: Exception) {
-            log.warn("Failed to send device token message immediately, will be retried by scheduler", e)
-        } finally {
-            if (map.isNullOrEmpty()) {
-                outboxIdsByEventIdentity.remove()
-            }
-        }
     }
 
     private fun createMessage(event: DeviceTokenEvent): Pair<String, Any> {

dms-main/main-infrastructure/src/main/kotlin/team/aliens/dms/event/handler/NotificationEventHandler.kt

@@ -1,7 +1,6 @@
 package team.aliens.dms.event.handler
 
 import com.fasterxml.jackson.databind.ObjectMapper
-import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Component
 import org.springframework.transaction.event.TransactionPhase
 import org.springframework.transaction.event.TransactionalEventListener
@@ -15,23 +14,18 @@ import team.aliens.dms.event.GroupNotificationEvent
 import team.aliens.dms.event.NotificationEvent
 import team.aliens.dms.event.SingleNotificationEvent
 import team.aliens.dms.event.TopicNotificationEvent
-import team.aliens.dms.thirdparty.messagebroker.NotificationProducer
-import java.util.UUID
 
 @Component
 class NotificationEventHandler(
     private val outboxPort: OutboxPort,
-    private val notificationProducer: NotificationProducer,
     private val objectMapper: ObjectMapper
 ) {
-    private val log = LoggerFactory.getLogger(this::class.java)
-    private val outboxIdsByEventIdentity = ThreadLocal<MutableMap<Int, UUID>>()
 
     @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
     fun saveOutbox(event: NotificationEvent) {
         val (eventType, message) = createMessage(event)
 
-        val saved = outboxPort.save(
+        outboxPort.save(
             OutboxData(
                 id = null,
                 aggregateType = "notification",
@@ -40,30 +34,6 @@ class NotificationEventHandler(
                 status = OutboxStatus.PENDING
             )
         )
-
-        val map = outboxIdsByEventIdentity.get() ?: mutableMapOf<Int, UUID>().also {
-            outboxIdsByEventIdentity.set(it)
-        }
-
-        map[System.identityHashCode(event)] = saved.id!!
-    }
-
-    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
-    fun publishMessage(event: NotificationEvent) {
-        val (_, message) = createMessage(event)
-        val map = outboxIdsByEventIdentity.get()
-        val outboxId = map.remove(System.identityHashCode(event))
-
-        try {
-            notificationProducer.sendMessage(message)
-            outboxId?.let { outboxPort.deleteById(it) }
-        } catch (e: Exception) {
-            log.warn("Failed to send notification immediately, will be retried by scheduler", e)
-        } finally {
-            if (map.isNullOrEmpty()) {
-                outboxIdsByEventIdentity.remove()
-            }
-        }
     }
 
     private fun createMessage(event: NotificationEvent): Pair<String, Any> {

dms-main/main-infrastructure/src/main/resources/application.yml

@@ -53,6 +53,9 @@ spring:
     job-store-type: jdbc
     jdbc:
       initialize-schema: never
+    overwrite-existing-jobs: true
+    auto-startup: true
+    wait-for-jobs-to-complete-on-shutdown: true
     properties:
       org:
         quartz:
@@ -65,6 +68,9 @@ spring:
             tablePrefix: QRTZ_
             isClustered: ${QUARTZ_CLUSTERED:false}
             useProperties: false
+            misfireThreshold: 300000
+            acquireTriggersWithinLock: true
+            txIsolationLevelSerializable: false
           threadPool:
             class: org.quartz.simpl.SimpleThreadPool
             threadCount: 10

dms-main/main-persistence/src/main/kotlin/team/aliens/dms/persistence/outbox/OutboxPersistenceAdapter.kt

@@ -6,7 +6,6 @@ import team.aliens.dms.common.dto.OutboxStatus
 import team.aliens.dms.common.spi.OutboxPort
 import team.aliens.dms.persistence.outbox.mapper.OutboxMapper
 import team.aliens.dms.persistence.outbox.repository.OutboxJpaRepository
-import java.util.UUID
 
 @Component
 class OutboxPersistenceAdapter(
@@ -20,10 +19,6 @@ class OutboxPersistenceAdapter(
         return outboxMapper.toDomain(saved)
     }
 
-    override fun deleteById(id: UUID) {
-        outboxJpaRepository.deleteById(id)
-    }
-
     override fun findByStatus(status: OutboxStatus): List<OutboxData> {
         val jpaStatus = outboxMapper.toJpaStatus(status)
         return outboxJpaRepository.findByStatus(jpaStatus)