feat: transfer-relay 멀티 인스턴스가 아닌, 단일 스레드에서 multiThread 방식으로 개선

Author: hyungkishin

services/transfer/application/build.gradle.kts

@@ -1,30 +1,10 @@
-//plugins {
-//    kotlin("jvm")
-//    kotlin("plugin.spring")
-//}
-//
-//dependencies {
-//    implementation(project(":transfer-domain"))
-//    implementation(project(":common-application"))
-//    implementation(project(":common-domain"))
-//
-//    implementation("org.springframework:spring-context")
-//    implementation("org.springframework:spring-tx")
-//
-//    testImplementation("io.kotest:kotest-runner-junit5")
-//    testImplementation("io.kotest:kotest-assertions-core")
-//}
-
 plugins {
     id("transentia.spring-library")
 }
 
 dependencies {
-    // 프로젝트 의존성
     implementation(project(":transfer-domain"))
     implementation(project(":common-application"))
     implementation(project(":common-domain"))
-
-    // 특화된 의존성 (있다면 추가)
-    // 예: implementation("org.springframework.retry:spring-retry")
+    implementation(project(":kafka-model"))
 }

services/transfer/application/src/main/kotlin/io/github/hyungkishin/transentia/application/TransactionService.kt

@@ -1,66 +1,97 @@
 package io.github.hyungkishin.transentia.application
 
+import io.github.hyungkishin.transentia.application.mapper.OutboxEventMapper
 import io.github.hyungkishin.transentia.application.provided.TransactionRegister
 import io.github.hyungkishin.transentia.application.provided.command.TransferRequestCommand
 import io.github.hyungkishin.transentia.application.required.TransactionRepository
+import io.github.hyungkishin.transentia.application.required.TransferEventsOutboxRepository
 import io.github.hyungkishin.transentia.application.required.UserRepository
 import io.github.hyungkishin.transentia.application.required.command.TransferResponseCommand
 import io.github.hyungkishin.transentia.common.error.CommonError
 import io.github.hyungkishin.transentia.common.error.DomainException
+import io.github.hyungkishin.transentia.common.message.transfer.TransferCompleted
 import io.github.hyungkishin.transentia.common.snowflake.IdGenerator
 import io.github.hyungkishin.transentia.common.snowflake.SnowFlakeId
 import io.github.hyungkishin.transentia.container.model.transaction.Transaction
 import io.github.hyungkishin.transentia.container.validator.transfer.TransferValidator
+import org.slf4j.LoggerFactory
 import org.springframework.context.ApplicationEventPublisher
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
+import java.time.Instant
 
 @Service
 class TransactionService(
     private val transactionRepository: TransactionRepository,
     private val userRepository: UserRepository,
-    private val transactionHistoryService: TransactionHistoryService,
+    private val outboxRepository: TransferEventsOutboxRepository,
+    private val outboxEventMapper: OutboxEventMapper,
     private val idGenerator: IdGenerator,
     private val eventPublisher: ApplicationEventPublisher,
 ) : TransactionRegister {
 
+    private val log = LoggerFactory.getLogger(javaClass)
+
     @Transactional
     override fun createTransfer(command: TransferRequestCommand): TransferResponseCommand {
-        val sender = userRepository.findById(command.senderId) ?: throw DomainException(
-            CommonError.NotFound("account_balance", command.senderId.toString()),
-            "송신자 정보를 찾을 수 없습니다. senderId=${command.senderId}"
-        )
-
-        val receiver = userRepository.findByAccountNumber(command.receiverAccountNumber) ?: throw DomainException(
-            CommonError.NotFound("account_balance", command.receiverAccountNumber.toString()),
-            "수신자 계좌 정보를 찾을 수 없습니다. snowFlakeId=${command.receiverAccountNumber}"
-        )
+        val (sender, receiver) = loadUsers(command)
+        val amount = command.amount()
 
-        // TODO: - 테스트의 용이성과 확장성 / 재사용성 검증하기
-        TransferValidator.validate(sender, receiver, command.amount())
+        TransferValidator.validate(sender, receiver, amount)
 
         val transaction = Transaction.of(
             SnowFlakeId(idGenerator.nextId()),
             sender.id,
             receiver.id,
-            command.amount()
+            amount
         )
 
-        sender.accountBalance.withdrawOrThrow(command.amount())
-        receiver.accountBalance.deposit(command.amount())
-
+        sender.accountBalance.withdrawOrThrow(amount)
+        receiver.accountBalance.deposit(amount)
         userRepository.save(sender)
         userRepository.save(receiver)
 
-        val completeEvent = transaction.complete()
         val savedTransaction = transactionRepository.save(transaction)
 
-        // TODO: outbox ( kafka publish ) + relay 서버를 fadeout 하고, CDC 방식으로 전환.
+        val completeEvent = transaction.complete()
+
+        // outbox 먼저 저장
+        saveToOutbox(completeEvent, savedTransaction.id.value)
+
+        // 이벤트 발행 (커밋 후 별도 스레드에서 Kafka 전송)
         eventPublisher.publishEvent(completeEvent)
 
         return TransferResponseCommand.from(savedTransaction)
     }
 
+    private fun saveToOutbox(event: TransferCompleted, transactionId: Long) {
+        try {
+            val outboxEvent = outboxEventMapper.toOutboxEvent(event, transactionId)
+            outboxRepository.save(outboxEvent, Instant.now())
+        } catch (e: Exception) {
+            throw DomainException(
+                CommonError.Conflict("outbox_save_failed"),
+                "송금 처리 중 시스템 오류가 발생했습니다.",
+                e
+            )
+        }
+    }
+
+    private fun loadUsers(command: TransferRequestCommand) =
+        Pair(
+            userRepository.findById(command.senderId)
+                ?: throw DomainException(
+                    CommonError.NotFound("account_balance", command.senderId.toString()),
+                    "송신자 정보를 찾을 수 없습니다."
+                ),
+            userRepository.findByAccountNumber(command.receiverAccountNumber)
+                ?: throw DomainException(
+                    CommonError.NotFound("account_balance", command.receiverAccountNumber),
+                    "수신자 계좌 정보를 찾을 수 없습니다."
+                )
+        )
+
+    @Transactional(readOnly = true)
     override fun findTransfer(transactionId: Long): TransferResponseCommand {
         val tx = transactionRepository.findById(transactionId)
             ?: throw DomainException(
@@ -69,5 +100,4 @@ class TransactionService(
             )
         return TransferResponseCommand.from(tx)
     }
-
-}
+}

services/transfer/application/src/main/kotlin/io/github/hyungkishin/transentia/application/handler/TransferOutboxEventHandler.kt

@@ -1,6 +1,6 @@
 package io.github.hyungkishin.transentia.application.handler
 
-import io.github.hyungkishin.transentia.application.required.HybridFdsEventPublisher
+import io.github.hyungkishin.transentia.application.port.TransferEventPublisher
 import io.github.hyungkishin.transentia.common.message.transfer.TransferCompleted
 import org.slf4j.LoggerFactory
 import org.springframework.scheduling.annotation.Async
@@ -10,31 +10,16 @@ import org.springframework.transaction.event.TransactionalEventListener
 
 @Component
 class TransferOutboxEventHandler(
-    private val hybridFdsEventPublisher: HybridFdsEventPublisher
+    private val eventPublisher: TransferEventPublisher
 ) {
-
     private val log = LoggerFactory.getLogger(javaClass)
 
     @Async("outboxEventExecutor")
     @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
     fun handle(event: TransferCompleted) {
+        log.debug("비동기 Kafka 전송 시도: transactionId={}, eventId={}", event.transactionId, event.eventId)
 
-        val currentThread = Thread.currentThread()
-
-        val threadConfigData = mapOf(
-            "threadName" to currentThread.name,
-            "threadGroup" to (currentThread.threadGroup?.name ?: "N/A"),
-            "threadId" to currentThread.id.toString(),
-            "isDaemon" to currentThread.isDaemon.toString()
-        )
-
-        println("threadConfigData: $threadConfigData")
-
-        val kafkaSuccess = hybridFdsEventPublisher.publish(event)
-
-        if (!kafkaSuccess) {
-            log.warn("Kafka 즉시 전송 실패, Outbox에 저장됨: transactionId={}", event.transactionId)
-        }
+        eventPublisher.publish(event)
     }
 
-}
+}

services/transfer/application/src/main/kotlin/io/github/hyungkishin/transentia/application/mapper/OutboxEventMapper.kt

@@ -0,0 +1,41 @@
+package io.github.hyungkishin.transentia.application.mapper
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.github.hyungkishin.transentia.common.message.transfer.TransferCompleted
+import io.github.hyungkishin.transentia.container.event.TransferEvent
+import org.slf4j.MDC
+import org.springframework.stereotype.Component
+import java.util.*
+
+@Component
+class OutboxEventMapper(
+    private val objectMapper: ObjectMapper
+) {
+    fun toOutboxEvent(event: TransferCompleted, transactionId: Long): TransferEvent {
+        return TransferEvent(
+            eventId = event.eventId,
+            aggregateType = "Transaction",
+            aggregateId = transactionId.toString(),
+            eventType = "TRANSFER_COMPLETED",
+            payload = objectMapper.writeValueAsString(
+                mapOf(
+                    "transactionId" to event.transactionId,
+                    "senderId" to event.senderUserId,
+                    "receiverId" to event.receiverUserId,
+                    "amount" to event.amount,
+                    "status" to "COMPLETED",
+                    "occurredAt" to event.occurredAt.toEpochMilli()
+                )
+            ),
+            headers = objectMapper.writeValueAsString(
+                mapOf(
+                    "eventType" to "TRANSFER_COMPLETED",
+                    "eventVersion" to "v1",
+                    "traceId" to (MDC.get("traceId") ?: UUID.randomUUID().toString()),
+                    "producer" to "transfer-api",
+                    "contentType" to "application/json"
+                )
+            )
+        )
+    }
+}

services/transfer/application/src/main/kotlin/io/github/hyungkishin/transentia/application/port/TransferEventPublisher.kt

@@ -0,0 +1,15 @@
+package io.github.hyungkishin.transentia.application.port
+
+import io.github.hyungkishin.transentia.common.message.transfer.TransferCompleted
+
+/**
+ * 송금 이벤트 발행 Port
+ */
+interface TransferEventPublisher {
+    /**
+     * 송금 완료 이벤트 발행
+     * 
+     * 호출자가 이미 비동기 스레드에서 실행 중
+     */
+    fun publish(event: TransferCompleted)
+}

services/transfer/application/src/main/kotlin/io/github/hyungkishin/transentia/application/required/TransferEventsOutboxRepository.kt

@@ -9,33 +9,18 @@ interface TransferEventsOutboxRepository {
     fun save(row: TransferEvent, now: Instant)
 
     /**
-     * 처리할 이벤트들을 배치로 조회하고 SENDING 상태로 변경한다.
+     * 처리 대기 중인 이벤트를 조회하고 SENDING 상태로 변경
      *
-     * 여러 스레드나 프로세스가 동시에 실행되어도 안전하도록 SKIP LOCKED를 사용한다.
-     * Stuck SENDING 상태(stuckThresholdSeconds 이상 진행 중)인 이벤트도 자동으로 복구하여 처리한다.
-     * 우선순위는 PENDING > SENDING(Stuck) > FAILED 순으로 처리한다.
-     *
-     * @param limit 한 번에 처리할 최대 이벤트 수
-     * @param now 기준 시간 (기본값: 현재 시간, 테스트 시 고정 시간 주입 가능)
-     * @param stuckThresholdSeconds Stuck SENDING 판단 기준 시간 (초)
-     * @return 처리할 이벤트 목록
+     * SKIP LOCKED로 동시성 제어
+     * 우선순위: PENDING > SENDING(Stuck) > FAILED
      */
     fun claimBatch(
         limit: Int,
         now: Instant,
-        stuckThresholdSeconds: Long = 600
+        sendingTimeoutSeconds: Long = 120
     ): List<ClaimedRow>
 
-    fun markAsPublished(
-        ids: List<Long>,
-        now: Instant,
-    )
-
-    fun markFailedWithBackoff(
-        id: Long,
-        cause: String?,
-        backoffMillis: Long,
-        now: Instant,
-    )
+    fun markAsPublished(ids: List<Long>, now: Instant)
 
+    fun markFailedWithBackoff(id: Long, cause: String?, backoffMillis: Long, now: Instant)
 }

services/transfer/infra/src/main/kotlin/io/github/hyungkishin/transentia/infra/adapter/KafkaTransferEventPublisher.kt

@@ -0,0 +1,63 @@
+package io.github.hyungkishin.transentia.infra.adapter
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.github.hyungkishin.transentia.application.port.TransferEventPublisher
+import io.github.hyungkishin.transentia.application.required.TransferEventsOutboxRepository
+import io.github.hyungkishin.transentia.common.message.transfer.TransferCompleted
+import io.github.hyungkishin.transentia.infrastructure.kafka.model.TransferEventAvroModel
+import io.github.hyungkishin.transentia.infrastructure.kafka.model.TransferEventType
+import io.github.hyungkishin.transentia.infrastructure.kafka.model.TransferStatus
+import io.github.hyungkishin.transentia.infrastructure.kafka.producer.service.KafkaProducer
+import org.slf4j.LoggerFactory
+import org.slf4j.MDC
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.stereotype.Component
+import java.time.Instant
+import java.util.*
+
+@Component
+class KafkaTransferEventPublisher(
+    private val kafkaProducer: KafkaProducer<String, TransferEventAvroModel>,
+    private val outboxRepository: TransferEventsOutboxRepository,
+    private val objectMapper: ObjectMapper,
+    @Value("\${app.kafka.topics.transfer-events}") private val topicName: String
+) : TransferEventPublisher {
+
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    override fun publish(event: TransferCompleted) {
+        try {
+            val avroModel = TransferEventAvroModel.newBuilder()
+                .setEventId(event.eventId)
+                .setEventType(TransferEventType.TRANSFER_COMPLETED)
+                .setAggregateId(event.transactionId.toString())
+                .setTransactionId(event.transactionId)
+                .setSenderId(event.senderUserId)
+                .setReceiverId(event.receiverUserId)
+                .setAmount(event.amount.toString())
+                .setStatus(TransferStatus.COMPLETED)
+                .setOccurredAt(event.occurredAt.toEpochMilli())
+                .setHeaders(
+                    objectMapper.writeValueAsString(
+                        mapOf(
+                            "eventType" to "TRANSFER_COMPLETED",
+                            "eventVersion" to "v1",
+                            "traceId" to (MDC.get("traceId") ?: UUID.randomUUID().toString()),
+                            "producer" to "transfer-api",
+                            "contentType" to "application/json"
+                        )
+                    )
+                )
+                .setCreatedAt(System.currentTimeMillis())
+                .build()
+
+            kafkaProducer.sendSync(topicName, avroModel)
+            
+            outboxRepository.markAsPublished(listOf(event.eventId), Instant.now())
+            log.debug("Kafka 전송 및 outbox PUBLISHED 완료: eventId={}", event.eventId)
+            
+        } catch (e: Exception) {
+            log.warn("Kafka 전송 실패 (relay 재시도): eventId={}, error={}", event.eventId, e.message)
+        }
+    }
+}