feature/33-kafka

buildSrc/src/main/kotlin/Dependencies.kt

@@ -50,5 +50,7 @@ object Dependencies {
     const val SENTRY_SPRING_BOOT_STARTER = "io.sentry:sentry-spring-boot-starter-jakarta:${DependencyVersion.SENTRY}"
 
     // Spring Cloud Config
-    const val SPRING_CLOUD_STARTER_CONFIG = "org.springframework.cloud:spring-cloud-starter-config"
+    const val SPRING_CLOUD_STARTER_CONFIG = "org.springframework.cloud:spring-cloud-starter:2024.0.2"
+
+    const val KAFKA = "org.springframework.kafka:spring-kafka"
 }

buildSrc/src/main/kotlin/Plugin.kt

@@ -1,6 +1,7 @@
 object Plugin {
     const val KOTLIN_JVM = "org.jetbrains.kotlin.jvm"
     const val KOTLIN_SPRING = "org.jetbrains.kotlin.plugin.spring"
+    const val KOTLIN_JPA = "org.jetbrains.kotlin.plugin.jpa"
     const val KOTLIN_KAPT = "org.jetbrains.kotlin.kapt"
     const val SPRING_BOOT = "org.springframework.boot"
     const val SPRING_DEPENDENCY_MANAGEMENT = "io.spring.dependency-management"

casper-user/build.gradle.kts

@@ -4,6 +4,7 @@ import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 plugins {
     id(Plugin.KOTLIN_JVM) version PluginVersion.KOTLIN_VERSION
     id(Plugin.KOTLIN_SPRING) version PluginVersion.KOTLIN_VERSION
+    id(Plugin.KOTLIN_JPA) version PluginVersion.KOTLIN_VERSION
     id(Plugin.KOTLIN_KAPT)
     id(Plugin.SPRING_BOOT) version PluginVersion.SPRING_BOOT_VERSION
     id(Plugin.SPRING_DEPENDENCY_MANAGEMENT) version PluginVersion.SPRING_DEPENDENCY_MANAGEMENT_VERSION
@@ -78,8 +79,11 @@ dependencies {
     // Sentry
     implementation(Dependencies.SENTRY_SPRING_BOOT_STARTER)
 
+    //kafka
+    implementation(Dependencies.KAFKA)
+
     // Spring Cloud Config
-    implementation(Dependencies.SPRING_CLOUD_STARTER_CONFIG)
+    //implementation(Dependencies.SPRING_CLOUD_STARTER_CONFIG)
 }
 
 protobuf {

casper-user/src/main/kotlin/hs/kr/entrydsm/user/domain/user/adapter/out/persistence/UserPersistenceAdapter.kt

@@ -81,6 +81,14 @@ class UserPersistenceAdapter(
         userRepository.deleteById(userId)
     }
 
+    /**
+     * 모든 사용자를 삭제합니다.
+     * 관리자의 전체 데이터 초기화 시에만 사용됩니다.
+     */
+    override fun deleteAll() {
+        userRepository.deleteAll()
+    }
+
     /**
      * 지정된 일수보다 오래된 탈퇴 사용자를 조회합니다.
      *

casper-user/src/main/kotlin/hs/kr/entrydsm/user/domain/user/application/port/out/DeleteUserPort.kt

@@ -22,4 +22,10 @@ interface DeleteUserPort {
      * @return 삭제 대상 사용자 목록
      */
     fun findWithdrawnUsersOlderThan(days: Long): List<User>
+
+    /**
+     * 모든 사용자를 삭제합니다.
+     * 관리자의 전체 데이터 초기화 시에만 사용됩니다.
+     */
+    fun deleteAll()
 }

casper-user/src/main/kotlin/hs/kr/entrydsm/user/domain/user/application/service/ChangeReceiptCodeService.kt

@@ -4,6 +4,7 @@ import hs.kr.entrydsm.user.domain.user.application.port.`in`.ChangeReceiptCodeUs
 import hs.kr.entrydsm.user.domain.user.application.port.out.QueryUserPort
 import hs.kr.entrydsm.user.domain.user.application.port.out.SaveUserPort
 import hs.kr.entrydsm.user.domain.user.exception.UserNotFoundException
+import hs.kr.entrydsm.user.infrastructure.kafka.producer.UserEventProducer
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
 import java.util.UUID
@@ -14,12 +15,13 @@ import java.util.UUID
  *
  * @property queryUserPort 사용자 조회 포트
  * @property saveUserPort 사용자 저장 포트
+ * @property userEventProducer 사용자 이벤트 발행기
  */
-@Transactional
 @Service
 class ChangeReceiptCodeService(
     private val queryUserPort: QueryUserPort,
     private val saveUserPort: SaveUserPort,
+    private val userEventProducer: UserEventProducer
 ) : ChangeReceiptCodeUseCase {
     /**
      * 사용자의 접수코드를 변경합니다.
@@ -28,12 +30,39 @@ class ChangeReceiptCodeService(
      * @param receiptCode 새로운 접수코드
      * @throws UserNotFoundException 사용자가 존재하지 않는 경우
      */
+    @Transactional
     override fun changeReceiptCode(
         userId: UUID,
         receiptCode: Long,
     ) {
-        val user = queryUserPort.findById(userId) ?: throw UserNotFoundException
-        val updateUser = user.changeReceiptCode(receiptCode)
-        saveUserPort.save(updateUser)
+        try {
+            val user = queryUserPort.findById(userId)
+
+            if (user == null) {
+                userEventProducer.sendReceiptCodeUpdateFailed(
+                    receiptCode = receiptCode,
+                    userId = userId,
+                    reason = "User not found"
+                )
+                throw UserNotFoundException
+            }
+
+            val updatedUser = user.copy(receiptCode = receiptCode)
+            saveUserPort.save(updatedUser)
+
+            // 성공 이벤트 발행
+            userEventProducer.sendReceiptCodeUpdateCompleted(receiptCode, userId)
+
+        } catch (e: Exception) {
+
+            if (e !is UserNotFoundException) {
+                userEventProducer.sendReceiptCodeUpdateFailed(
+                    receiptCode = receiptCode,
+                    userId = userId,
+                    reason = e.message ?: "Unknown error"
+                )
+            }
+            throw e  // 예외 다시 던져서 롤백 발생
+        }
     }
 }

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/configuration/KafkaConsumerConfig.kt

@@ -0,0 +1,45 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.configuration
+
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.kafka.annotation.EnableKafka
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory
+import org.springframework.kafka.support.serializer.JsonDeserializer
+
+@EnableKafka
+@Configuration
+class KafkaConsumerConfig(
+    private val kafkaProperty: KafkaProperty
+) {
+
+    @Bean
+    fun kafkaListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, String> {
+        return ConcurrentKafkaListenerContainerFactory<String, String>().apply {
+            setConcurrency(2)
+            consumerFactory = DefaultKafkaConsumerFactory(consumerFactoryConfig())
+            containerProperties.pollTimeout = 500
+        }
+    }
+
+    private fun consumerFactoryConfig(): Map<String, Any> {
+        return mapOf(
+            ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaProperty.serverAddress,
+            ConsumerConfig.ISOLATION_LEVEL_CONFIG to "read_committed",
+            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
+            ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to "false",
+            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "latest",
+            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to JsonDeserializer::class.java,
+            ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG to 5000,
+            JsonDeserializer.TRUSTED_PACKAGES to "*",
+            "security.protocol" to "SASL_PLAINTEXT",
+            "sasl.mechanism" to "SCRAM-SHA-512",
+            "sasl.jaas.config" to
+                    "org.apache.kafka.common.security.scram.ScramLoginModule required " +
+                    "username=\"${kafkaProperty.confluentApiKey}\" " +
+                    "password=\"${kafkaProperty.confluentApiSecret}\";"
+        )
+    }
+}

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/configuration/KafkaProducerConfig.kt

@@ -0,0 +1,60 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.configuration
+
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.common.serialization.StringSerializer
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.kafka.core.DefaultKafkaProducerFactory
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.kafka.support.serializer.JsonSerializer
+
+@Configuration
+class KafkaProducerConfig(
+    private val kafkaProperty: KafkaProperty
+) {
+
+    @Bean
+    fun deleteAllTableProducerFactory(): DefaultKafkaProducerFactory<String, Unit> {
+        return DefaultKafkaProducerFactory(producerConfig())
+    }
+
+    @Bean
+    fun deleteAllTableKafkaTemplate(): KafkaTemplate<String, Unit> {
+        return KafkaTemplate(deleteAllTableProducerFactory())
+    }
+
+    @Bean
+    fun deleteUserProducerFactory(): DefaultKafkaProducerFactory<String, Long> {
+        return DefaultKafkaProducerFactory(producerConfig())
+    }
+
+    @Bean
+    fun deleteUserKafkaTemplate(): KafkaTemplate<String, Long> {
+        return KafkaTemplate(deleteUserProducerFactory())
+    }
+
+    @Bean
+    fun userEventProducerFactory(): DefaultKafkaProducerFactory<String, Any> {
+        return DefaultKafkaProducerFactory(producerConfig())
+    }
+
+    @Bean
+    fun userEventKafkaTemplate(): KafkaTemplate<String, Any> {
+        return KafkaTemplate(userEventProducerFactory())
+    }
+
+
+    private fun producerConfig(): Map<String, Any> {
+        return mapOf(
+            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaProperty.serverAddress,
+            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
+            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to JsonSerializer::class.java,
+            "security.protocol" to "SASL_PLAINTEXT",
+            "sasl.mechanism" to "SCRAM-SHA-512",
+            "sasl.jaas.config" to
+                    "org.apache.kafka.common.security.scram.ScramLoginModule required " +
+                    "username=\"${kafkaProperty.confluentApiKey}\" " +
+                    "password=\"${kafkaProperty.confluentApiSecret}\";"
+        )
+    }
+}

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/configuration/KafkaProperty.kt

@@ -0,0 +1,12 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.configuration
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+import org.springframework.boot.context.properties.ConfigurationPropertiesBinding
+
+@ConfigurationPropertiesBinding
+@ConfigurationProperties("kafka")
+class KafkaProperty(
+    val serverAddress: String,
+    val confluentApiKey: String,
+    val confluentApiSecret: String
+)

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/configuration/KafkaTopics.kt

@@ -0,0 +1,11 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.configuration
+
+object KafkaTopics {
+    const val DELETE_USER = "delete-user"
+    const val DELETE_ALL_TABLE = "delete-all-table"
+    const val CREATE_APPLICATION = "create-application"
+
+    // Choreography 이벤트들
+    const val USER_RECEIPT_CODE_UPDATE_COMPLETED = "user-receipt-code-update-completed"
+    const val USER_RECEIPT_CODE_UPDATE_FAILED = "user-receipt-code-update-failed"
+}

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/consumer/CreateApplicationConsumer.kt

@@ -0,0 +1,25 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.consumer
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import hs.kr.entrydsm.user.domain.user.application.port.`in`.ChangeReceiptCodeUseCase
+import hs.kr.entrydsm.user.infrastructure.kafka.configuration.KafkaTopics
+import hs.kr.entrydsm.user.infrastructure.kafka.consumer.dto.CreateApplicationEvent
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.stereotype.Service
+
+@Service
+class CreateApplicationConsumer(
+    private val changeReceiptCodeUseCase: ChangeReceiptCodeUseCase,
+    private val mapper: ObjectMapper
+) {
+    @KafkaListener(
+        topics = [KafkaTopics.CREATE_APPLICATION],
+        groupId = "change-user-receipt-code-consumer",
+        containerFactory = "kafkaListenerContainerFactory"
+    )
+    fun execute(message: String) {
+        val createApplicationEvent = mapper.readValue(message, CreateApplicationEvent::class.java)
+        changeReceiptCodeUseCase.changeReceiptCode(createApplicationEvent.userId, createApplicationEvent.receiptCode)
+    }
+
+}

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/consumer/DeleteUserTableConsumer.kt

@@ -0,0 +1,19 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.consumer
+
+import hs.kr.entrydsm.user.domain.user.application.port.out.DeleteUserPort
+import hs.kr.entrydsm.user.infrastructure.kafka.configuration.KafkaTopics
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.transaction.annotation.Transactional
+
+open class DeleteUserTableConsumer(
+    private val deleteUserPort: DeleteUserPort
+) {
+    @KafkaListener(
+        topics = [KafkaTopics.DELETE_ALL_TABLE],
+        groupId = "delete-all-table-user",
+        containerFactory = "kafkaListenerContainerFactory"
+    )
+    @Transactional
+    open fun execute() = deleteUserPort.deleteAll()
+
+}

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/consumer/dto/CreateApplicationEvent.kt

@@ -0,0 +1,8 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.consumer.dto
+
+import java.util.UUID
+
+data class CreateApplicationEvent(
+    val receiptCode: Long,
+    val userId: UUID
+)

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/consumer/dto/UserReceiptCodeUpdateCompletedEvent.kt

@@ -0,0 +1,8 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.consumer.dto
+
+import java.util.UUID
+
+data class UserReceiptCodeUpdateCompletedEvent(
+    val receiptCode: Long,
+    val userId: UUID
+)

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/consumer/dto/UserReceiptCodeUpdateFailedEvent.kt

@@ -0,0 +1,9 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.consumer.dto
+
+import java.util.UUID
+
+data class UserReceiptCodeUpdateFailedEvent(
+    val receiptCode: Long,
+    val userId: UUID,
+    val reason: String
+)

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/producer/DeleteAllTableProducerImpl.kt

@@ -0,0 +1,17 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.producer
+
+import hs.kr.entrydsm.user.infrastructure.kafka.configuration.KafkaTopics
+import org.springframework.context.annotation.Profile
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.stereotype.Component
+
+
+@Component
+class DeleteAllTableProducerImpl(
+    private val kafkaTemplate: KafkaTemplate<String, Unit>,
+) : DeleteAllTableProducer {
+
+    override fun send() {
+        kafkaTemplate.send(KafkaTopics.DELETE_ALL_TABLE, Unit)
+    }
+}

casper-user/src/main/kotlin/hs/kr/entrydsm/user/infrastructure/kafka/producer/DeleteUserProducerImpl.kt

@@ -0,0 +1,15 @@
+package hs.kr.entrydsm.user.infrastructure.kafka.producer
+
+import hs.kr.entrydsm.user.infrastructure.kafka.configuration.KafkaTopics
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.stereotype.Component
+
+
+@Component
+class DeleteUserProducerImpl(
+    private val kafkaTemplate: KafkaTemplate<String, Long>
+) : DeleteUserProducer {
+    override fun send(receiptCode: Long) {
+        kafkaTemplate.send(KafkaTopics.DELETE_USER, receiptCode)
+    }
+}