content: package io.cosmos.configuration;
  
  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 java.util.HashMap;
  import java.util.Map;
  
  @EnableKafka
  @Configuration
  public class KafkaConfiguration {
      //
      @Value("${spring.application.name}")
      private String serviceName;
      @Value("${broker.kafka}")
      private String kafkaHost;
  
      @Bean
      public ConsumerFactory<String, String> consumerFactory() {
          //
          Map<String, Object> props = new HashMap<>();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaHost);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, serviceName);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          return new DefaultKafkaConsumerFactory<>(props);
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
          //
          ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  
      @Bean
      public ProducerFactory<String, String> producerFactory() {
          //
          Map<String, Object> configProps = new HashMap<>();
          configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaHost);
          configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          return new DefaultKafkaProducerFactory<>(configProps);
      }
  
      @Bean
      public KafkaTemplate<String, String> kafkaTemplate() {
          //
          return new KafkaTemplate<>(producerFactory());
      }
  }
  