content: package shop.kkeujeok.kkeujeokbackend.global.config;
  
  import java.util.HashMap;
  import java.util.Map;
  import org.apache.kafka.clients.admin.NewTopic;
  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.config.ConcurrentKafkaListenerContainerFactory;
  import org.springframework.kafka.core.ConsumerFactory;
  import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  import org.springframework.kafka.core.KafkaTemplate;
  import org.springframework.kafka.core.ProducerFactory;
  
  @Configuration
  public class KafkaConfig {
  
      @Value("${kafka.bootstrap-servers}")
      private String bootstrapServers;
  
      @Value("${kafka.topic-name}")
      private String topicName;
  
      @Value("${kafka.group-id}")
      private String groupId;
  
      @Bean
      public NewTopic topic() {
          return new NewTopic(topicName, 3, (short) 1);
      }
  
      // Producer 설정
      @Bean
      public ProducerFactory<String, String> producerFactory() {
          Map<String, Object> config = new HashMap<>();
          config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
          config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          return new DefaultKafkaProducerFactory<>(config);
      }
  
      @Bean
      public KafkaTemplate<String, String> kafkaTemplate() {
          return new KafkaTemplate<>(producerFactory());
      }
  
      // Consumer 설정
      @Bean
      public ConsumerFactory<String, String> consumerFactory() {
          Map<String, Object> config = new HashMap<>();
          config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
          config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
          config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          return new DefaultKafkaConsumerFactory<>(config);
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory<String, String> factory =
                  new ConcurrentKafkaListenerContainerFactory<>();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  }
  
  