(Kotlin-Kafka) Kotlin Spring Boot Kafka 통합 초기 세팅

dependencies {
    implementation("org.springframework.kafka:spring-kafka")
    implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
    testImplementation("org.springframework.kafka:spring-kafka-test")
}

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kafka.common.serialization.StringSerializer
import org.springframework.beans.factory.annotation.Value
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.*
import org.springframework.kafka.support.serializer.JsonDeserializer
import org.springframework.kafka.support.serializer.JsonSerializer

@EnableKafka
@Configuration
class KafkaConfig {

    @Value("\${spring.kafka.bootstrap-servers}")
    private val BOOTSTRAP_SERVER: String? = null

    @Value("\${spring.kafka.consumer.group-id}")
    private val GROUP_ID: String? = null

    @Value("\${spring.kafka.consumer.auto-offset-reset}")
    private val AUTO_OFFSET_RESET: String? = null

    // Consumer 설정을 위한 Factory Bean을 정의합니다.
    // 여러 개의 Consumer 스레드를 설정하기 위해 setConcurrency를 사용하여 동시성을 3으로 설정합니다.
    @Bean
    fun kafkaListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, Any> {
        val factory = ConcurrentKafkaListenerContainerFactory<String, Any>()
        factory.consumerFactory = consumerFactory()
        factory.setConcurrency(3)
        return factory
    }

    // Consumer 설정
    @Bean
    fun consumerFactory(): ConsumerFactory<String, Any> {
        val deserializer = JsonDeserializer(Any::class.java)
        deserializer.addTrustedPackages("*")

        val config = mapOf(
            ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to BOOTSTRAP_SERVER,
            ConsumerConfig.GROUP_ID_CONFIG to GROUP_ID,
            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to deserializer,
            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to AUTO_OFFSET_RESET
        )
        return DefaultKafkaConsumerFactory(config, StringDeserializer(), deserializer)
    }

    // Producer 설정을 위한 Factory Bean을 정의합니다.
    @Bean
    fun producerFactory(): ProducerFactory<String, Any> {
        val config = mapOf(
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to BOOTSTRAP_SERVER,
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to JsonSerializer::class.java
        )
        return DefaultKafkaProducerFactory(config)
    }

    @Bean
    fun kafkaTemplate(): KafkaTemplate<String, Any> {
        return KafkaTemplate(producerFactory())
    }
}

ipackage brn.neos.neos_board_backend.kafka.listener
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.springframework.kafka.annotation.KafkaListener
import org.springframework.stereotype.Service

@Service
class TopicUserListener() {
    val log = LoggerFactory.getLogger(TopicUserListener::class.java)

    // Kafka로부터 "user-join" 토픽을 구독하는 Consumer 메서드
    @KafkaListener(topics = ["user-join"])
    fun consumeConfirmUser(message: ConsumerRecord<String, Map<String, Any>>) {
        // 처리 로직
    }

    // Kafka로부터 "user-join-rollback" 토픽을 구독하는 Consumer 메서드
    @KafkaListener(topics = ["user-join-rollback"])
    fun consumeRollBackUser(message: ConsumerRecord<String, Map<String, Any>>) {
        // 처리 로직
    }
}