[SpringBoot] Kafka 연동(docker) 및 Consumer 구현하기

구현 환경

> Spring Boot 2.7.7

> Kotlin

---

Kafka 환경 설치 (Docker)

docker-compose.yml

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:2.12-2.5.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

아래 명령어를 사용하여 zookeeper / kafka 실행

docker-compose up -d

도커 대시보드 or 명령어를 이용하여 kafka 컨테이너 내부에 진입하여 아래 명령어로 토픽 생성

kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic test-topic

---

Kafka 설정

gradle 설정

dependencies {
	implementation("org.springframework.kafka:spring-kafka")
}

application.yml

kafka-consumer:
  config:
    kafka-test:
      bootstrap-servers: localhost:9092
      client-id: kafka-test-consumer
      group-id: kafka-test-consumer
      topic: test-topic

App Config 설정

• KafkaConsumerProperties 객체를 갖는 AppConfig 생성 (→ app config 파일이 불필요하다면 생략 가능)

@ConstructorBinding
@ConfigurationProperties
data class AppConfig(
    val kafkaConsumer: KafkaConsumerProperties
) {
	// 추후 아래와 같이 추가하여 config 파일로 사용할 수 있음
	companion object {
		const val SERVICE_NAME = "kafka-example"
	}
}

@SpringBootApplication
@EnableConfigurationProperties(AppConfig::class) // 이 부분 추가!
class KafkaApplication

fun main(args: Array<String>) {
	runApplication<KafkaApplication>(*args)
}

Kafka Consumer 관련 설정

KafkaConsumerConfig

• kafka consum을 위한 factory를 생성
• consumerFactory과 에러 처리를 위한 errorHandler도 지정

@Configuration
class KafkaConsumerConfig(
    private val kafkaConsumerErrorHandler: KafkaConsumerErrorHandler,
) {
    @Bean("testKafkaListenerContainerFactory")
    fun testKafkaListenerContainerFactory(
        appConfig: AppConfig
    ): ConcurrentKafkaListenerContainerFactory<String, String> =
        ConcurrentKafkaListenerContainerFactory<String, String>().apply {
            consumerFactory = appConfig.kafkaConsumer.consumerFactory(KAFKA_TEST)
            setCommonErrorHandler(kafkaConsumerErrorHandler)
        }

    companion object {
        const val KAFKA_TEST= "kafka-test"
    }
}

KafkaConsumerProperties

• kafka 관련 properties를 가지고 있는 객체
• https://kafka.apache.org/documentation/#consumerconfigs  

Apache Kafka

data class KafkaConsumerProperties(
    val config: Map<String, KafkaConsumerConfig>,
    val keyDeserializerClass: Class<*> = StringDeserializer::class.java,
    val valueDeserializerClass: Class<*> = StringDeserializer::class.java,
    val autoOffsetReset: String = "earliest" // earlist or lastest
) {
    data class KafkaConsumerConfig(
        val bootstrapServers: String,
        val clientId: String,
        val groupId: String,    // 필요시 사용
    )

    fun propertiesMap(kafkaConsumerConfig: KafkaConsumerConfig): Map<String, Any?> =
        mutableMapOf<String, Any?>(
            ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaConsumerConfig.bootstrapServers,
            ConsumerConfig.CLIENT_ID_CONFIG to kafkaConsumerConfig.clientId,
            ConsumerConfig.GROUP_ID_CONFIG to kafkaConsumerConfig.groupId,
            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to keyDeserializerClass,
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to valueDeserializerClass,
            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to autoOffsetReset,
        ).toMap()

    fun <V> consumerFactory(key: String): ConsumerFactory<String, V> {
        return DefaultKafkaConsumerFactory(propertiesMap(config[key]!!))
    }
}

---

Consumer 구현

@Component
@KafkaListener(
    topics = ["\\${kafka-consumer.config.kafka-test.topic}"],
    clientIdPrefix = "\\${kafka-consumer.config.kafka-test.client-id}",
    containerFactory = "testKafkaListenerContainerFactory"
)
class TestConsumer {
    @KafkaHandler
    fun listen(msg: String){
        println(msg)
    }
}

• application.yml에 정의한 topic을 가지고 오도록 지정
• clientIfPrefix (필수 x) : 이를 설정하지 않을 경우, 어플리케이션 내 동일한 clientId를 사용하는 곳이 있다면, InstanceAlreadyExistsException 을 발생 → 이를 방지하기 위해 각각 consumer마다 clientIdPrefix를 다르게 지정하는 것이 필요

---

실행 결과

Produce 명령어를 날려서 잘 consume하는지 확인

kafka-console-producer.sh --topic test-topic --broker-list localhost:9092
> this is a msg from doteloper

https://github.com/jeongum/kafka-test

GitHub - jeongum/kafka-test: kafka consumer example (springboot)