Kafka 사용하기 (Producer, Consumer, DLQ)

                                    0. kafka를 실행하기

1. Spring Boot Project와 Kafka를 연동하기

2. Front Server를 Producer로 등록하기

3. Noti Server를 Consumer로 등록하기

                                    implementation("org.apache.kafka:kafka-clients:3.9.1")

                                    @Configuration
class KafkaProducerConfig {

    @Bean
    fun kafkaProducer(
        @Value("\${spring.kafka.bootstrap-servers}") kafkaBootstrapServers: String,
        @Value("\${spring.kafka.producer.key-serializer}") keySerializer: String,
        @Value("\${spring.kafka.producer.value-serializer}") valueSerializer: String
    ): KafkaProducer<String, String> {
        val props = Properties().apply {
            put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers)
            put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer)
            put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer)
        }
        return KafkaProducer<String, String>(props)
    }
}

                                    spring:
  kafka:
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    bootstrap-servers: << Kafka 서버 주소 >>

                                    @Service
class KafkaProducerService(
    private val kafkaProducer: KafkaProducer<String, String>,
    private val objectMapper: ObjectMapper
) {
    private val logger = KotlinLogging.logger {}

    fun sendNoti(request: NotificationSaveRequest) {
        val topicName = "noti" // 예: Consumer가 구독 중인 동일 토픽
        try {
            // DTO -> JSON
            val jsonValue = objectMapper.writeValueAsString(request)

            // ProducerRecord 생성 후 send
            val record = ProducerRecord<String, String>(topicName, jsonValue)
            kafkaProducer.send(record) { metadata, exception ->
                if (exception == null) {
                    logger.info {
                        "Sent message to $topicName [offset=${metadata.offset()}, " +
                                "partition=${metadata.partition()}]" }
                } else {
                    logger.error { "Failed to send message: ${exception.message}" }
                }
            }
        } catch (e: Exception) {
            logger.error { "sendMessage error: ${e.message}" }
        }
    }
}

                                    ✍️  Spring 전용 Kafka module을 사용하시면
Object Mapper 없이 JsonSerializer를 직접 사용할 수 있습니다.

                                    🤔 왜 실패하는가?

* Kafka는 정상이라고 가정합니다. 우리가 대응 해야 하는 것은 Consumer 코드입니다.

1. Noti Server와 Kafka의 네트워크 오류
- 거의 발생하지 않지만 일시적으로 네트워크 오류로 인해 패킷이 손실 될 수 있습니다.

2. Noti Server의 DB오류
- Noti Server의 DB를 일시적으로 사용할 수 없다면, 해당 Noti는 저장되지 않습니다.
 저장되지 않은 Noti는 Client에게 제공될 수 없습니다.

3. Packet Parsing 오류
- Kafka Value로 저장된 데이터는 Byte[] 입니다. 이를 JSON으로 다시 Parsing 하기 위해
Noti Server에서 Object Mapper사용하여 변환을 시도할 것 입니다.

- 하지만 이때 잘못된 필드명이나 데이터 형식이 들어있다면, 오류가 발생할 수 있습니다.

                                    ✍️ solution
몇 번 retry 해보고 그래도 계속 안되면 더 이상 consume 진행에 의미가 없기에 나중에 다시 시도

                                    ✍️ solution
여기도 몇 번 retry 해보고 계속 안되면 더 이상 consume의 의미가 없으므로 consume 종료

                                    ✍️ solutions

Json Parsing Error가 발생하면 retry 없이 아래 3개의 방법 중 하나 이상을 선택한다.

1. log에 남긴다

2. 다른 queue에 남긴다

3. 다른 table에 남긴다

 데이터를 남긴 후 해당 메세지에 대한 commit을 진행한다. 치명적인 오류가 아니기 때문에 consume을
종료할 이유는 없다.

                                    @Service
class KafkaConsumerRunner(
    private val kafkaConsumer: KafkaConsumer<String, String>,
    private val objectMapper: ObjectMapper,
    private val notificationRepository: NotificationRepository
) {
    private val logger = KotlinLogging.logger {}
    private val kafkaDlqLogger = LoggerFactory.getLogger("com.example.notiApiServer.DLQ")

    @Volatile
    private var consumerRunning = true

    fun poll(vararg args: String?) {
        logger.info {"start consumer runner poll ()"}
        val topicName = "noti"
        kafkaConsumer.subscribe(listOf(topicName))

        logger.info {"consume starts"}

        // Graceful shutdown을 위한 shutdown hook 등록
        Runtime.getRuntime().addShutdownHook(Thread {
            println("Shutdown initiated.")
            turnOff()
        })

        while (consumerRunning) {
            val records = kafkaConsumer.poll(Duration.ofSeconds(3000))

            for (record in records) {
                logger.info {"record : ${record.topic()}, ${record.offset()}, ${record.value()}"}
                try {
                    val notiSaveRequest = recordToNotiSaveRequest(record)
                    // save in DB
                    val notification = Notification(
                        id = null,
                        publisherId = notiSaveRequest.publisherId,
                        receiverId = notiSaveRequest.receiverId,
                        notificationType = notiSaveRequest.notificationType,
                        targetBoardId = notiSaveRequest.targetBoardId,
                        createdAt = null
                    )
                    // save db
                    saveNoti(notification)

                    val topicPartition = TopicPartition(record.topic(), record.partition())
                    val offsetAndMetadata = OffsetAndMetadata(record.offset() + 1)
                    kafkaConsumer.commitSync(mapOf(topicPartition to offsetAndMetadata))

                    // commit offset !
                    logger.info { "Record with offset ${record.offset()} processed and committed." }
                } catch (e: JsonProcessingException) {
                    // skip this record
                    continue
                } catch (e: DataAccessException) {
                    logger.error {"DB error occurred!! at offset: ${record.offset()}"}
                }
            }
        }
    }

    private fun recordToNotiSaveRequest(record: ConsumerRecord<String, String>): NotificationSaveRequest {
        val topic = record.topic()

        val partition = record.partition()
        val offset = record.offset()
        val json = record.value()
        val key = Triple(topic, partition, offset)
        try {
            return objectMapper.readValue<NotificationSaveRequest>(json)
        } catch (e: JsonProcessingException) {
            logger.error {"toNotiSaveRequest parsing error"}
            // save log
            val errorDto = KafkaDlqRecord(
                topic = topic,
                value = json,
                offset = offset,
                revive = false,
                reviveDate = null
            )
            logKafkaDlqError(errorDto)
            // offset -> +1 enforce
            val topicPartition = TopicPartition(topic, partition)
            val offsetData = OffsetAndMetadata(offset + 1) // offset commit to next
            kafkaConsumer.commitSync(mapOf(topicPartition to offsetData))
            throw e
        }
    }

    private fun logKafkaDlqError(dlqRecord: KafkaDlqRecord) {
        try {
            MDC.put("topic", dlqRecord.topic)
            MDC.put("offset", dlqRecord.offset.toString())
            MDC.put("value", dlqRecord.value)
            MDC.put("revive", dlqRecord.revive.toString())
            MDC.put("reviveDate", dlqRecord.reviveDate.toString())
            kafkaDlqLogger.error("Kafka DLQ error occurred: {}", dlqRecord)
        } finally {
            MDC.clear()
        }
    }

    private fun saveNoti(noti: Notification) {
        try {
            notificationRepository.save(noti)
        } catch (e: DataAccessException) {
            throw e
        }
    }

    private fun turnOff() {
        consumerRunning = false
        kafkaConsumer.close()
    }

    private fun turnOn() {
        // scheduling?
    }
}

                                    Producer에서 Noti를 가지고 있다가 Kafka가 복구되면 다시 발행을 시도한다.