Spring Kafka에서 최신 메시지만 소비하는 Pub/Sub 방식 설정

package com.example.demo.global.config  
        
import org.apache.kafka.clients.consumer.Consumer  
import org.apache.kafka.common.TopicPartition  
import org.springframework.kafka.listener.ConsumerAwareRebalanceListener  
import org.springframework.stereotype.Component  
        
@Component  
class SeekToEndRebalanceListener: ConsumerAwareRebalanceListener {  
  override fun onPartitionsAssigned(consumer: Consumer<*, *>, partitions: MutableCollection<TopicPartition>) {  
      // 파티션 할당 시, 해당 파티션의 오프셋을 끝으로 이동  
      partitions.forEach { partition ->  
          consumer.seekToEnd(listOf(partition))  
      }  
  }  
}  

package com.example.demo.global.config  
        
import org.apache.kafka.clients.consumer.ConsumerConfig  
import org.apache.kafka.common.serialization.StringDeserializer  
import org.springframework.beans.factory.annotation.Value  
import org.springframework.context.annotation.Configuration  
import org.springframework.kafka.core.DefaultKafkaConsumerFactory  
import org.springframework.kafka.listener.ContainerProperties  
import org.springframework.kafka.listener.KafkaMessageListenerContainer  
import org.springframework.kafka.listener.MessageListener  
        
@Configuration  
class KafkaMessageListenerConfig(  
  private val seekToEndRebalanceListener: SeekToEndRebalanceListener,  
) {  
        
  @Value("\${spring.kafka.bootstrap-servers}")  
  lateinit var bootstrapServers: String  
        
  fun createKafkaMessageListenerContainer(  
      topic: String,  
      groupId: String,  
      messageListener: MessageListener<String, String>  
  ): KafkaMessageListenerContainer<String, String> {  
      val consumerProps = mapOf<String, Any>(  
          ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,  
          ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,  
          ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,  
          ConsumerConfig.GROUP_ID_CONFIG to groupId,  
          ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "latest"  
      )  
        
      val consumerFactory = DefaultKafkaConsumerFactory<String, String>(consumerProps)  
        
      // 구독할 토픽을 동적으로 지정  
      val containerProperties = ContainerProperties(topic)  
        
      // 메시지를 수동으로 처리할 Listener  
      containerProperties.messageListener = messageListener  
        
      // 컨슈머 리밸런싱이 일어날 때, 항상 최신 오프셋에서 읽기 시작  
      containerProperties.setConsumerRebalanceListener(seekToEndRebalanceListener)  
        
      return KafkaMessageListenerContainer(consumerFactory, containerProperties)  
  }  
        
}  

package com.example.demo.kafka.producer  
        
import com.example.demo.constant.KafkaTopic  
import org.springframework.kafka.core.KafkaTemplate  
import org.springframework.stereotype.Service  
        
        
@Service  
class UserProducer(  
  private val kafkaTemplate: KafkaTemplate<String, String>  
) {  
        
  fun send(message: String) {  
      kafkaTemplate.send(KafkaTopic.USER, message)  
  }  
        
}  

package com.example.demo.constant  
        
object KafkaTopic {  
  const val USER = "user"  
}  

package com.example.demo.constant  
        
object KafkaConsumerGroupId {  
  const val USER = "user"  
}  

package com.example.demo.global.config  
        
import jakarta.annotation.PostConstruct  
import org.springframework.context.annotation.Configuration  
import java.io.BufferedReader  
import java.io.InputStreamReader  
import java.util.UUID  
        
@Configuration  
class AppConfig {  
  var containerId = UUID.randomUUID().toString()  
        
  @PostConstruct  
  private fun initContainerId() {  
      try {  
          // 'hostname' 명령어 실행을 위한 ProcessBuilder 사용  
          val processBuilder = ProcessBuilder("hostname")  
          val process = processBuilder.start()  
          val reader = BufferedReader(InputStreamReader(process.inputStream))  
          val containerIdString = reader.readLine()  // hostname 명령어의 결과  
        
          // hostname 값이 비어 있지 않으면 containerId 업데이트  
          if (!containerIdString.isNullOrBlank()) {  
              containerId = containerIdString  
          }  
      } catch (e: Exception) {  
          // 예외 발생 시 아무 일도 일어나지 않음  
          println("Failed to retrieve container ID. Using default value.")  
          e.printStackTrace()  
      }  
  }  
        
}  

package com.example.demo.kafka.consumer  
        
import com.example.constant.KafkaConsumerGroupId  
import com.example.constant.KafkaTopic  
import com.example.global.config.AppConfig  
import com.example.global.config.KafkaMessageListenerConfig  
import org.springframework.context.annotation.Bean  
import org.springframework.kafka.listener.KafkaMessageListenerContainer  
import org.springframework.kafka.listener.MessageListener  
import org.springframework.stereotype.Service  
        
@Service  
class UserConsumer(  
  private val kafkaMessageListenerConfig: KafkaMessageListenerConfig,  
  private val appConfig: AppConfig,  
) {  
        
  @Bean  
  fun userConsume(): KafkaMessageListenerContainer<String, String> {  
      val container = kafkaMessageListenerConfig.createKafkaMessageListenerContainer(  
          topic = KafkaTopic.USER,  
          groupId = "${KafkaConsumerGroupId.USER}-${appConfig.containerId}",  
          messageListener = messageListener()  
      )  
      container.start()  
        
      return container  
  }  
        
  private fun messageListener(): MessageListener<String, String> {  
      return MessageListener { record ->  
                    
          println("record++" + record)  
      }  
  }  
        
}