content: package ru.amplicode.orders.config;
  
  import org.apache.kafka.clients.CommonClientConfigs;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.producer.ProducerConfig;
  import org.apache.kafka.common.config.SaslConfigs;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.apache.kafka.common.serialization.StringSerializer;
  import org.springframework.beans.factory.annotation.Value;
  import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
  import org.springframework.context.annotation.Bean;
  import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
  import org.springframework.kafka.config.KafkaListenerContainerFactory;
  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.listener.DefaultErrorHandler;
  import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
  import org.springframework.kafka.support.serializer.JsonDeserializer;
  import org.springframework.kafka.support.serializer.JsonSerializer;
  import org.springframework.util.backoff.FixedBackOff;
  import ru.amplicode.orders.rest.dto.OrderDto;
  import ru.amplicode.orders.service.dto.OrderDeliveryInfoDto;
  
  import java.util.Map;
  
  //@Configuration
  public class KafkaConfiguration {
  
      @Value("${kafka.username}")
      private String kafkaUsername;
  
      @Value("${kafka.password}")
      private String kafkaPassword;
  
      @Bean
      DefaultKafkaProducerFactory<String, String> stringProducerFactory(KafkaProperties properties) {
          Map<String, Object> producerProperties = properties.buildProducerProperties(null);
          producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  
          setSecurityProps(producerProperties);
  
          return new DefaultKafkaProducerFactory<>(producerProperties);
      }
  
      @Bean
      KafkaTemplate<String, String> stringKafkaTemplate(DefaultKafkaProducerFactory<String, String> stringProducerFactory) {
          return new KafkaTemplate<>(stringProducerFactory);
      }
  
      @Bean
      public ConsumerFactory<String, String> stringConsumerFactory(KafkaProperties kafkaProperties) {
          Map<String, Object> props = kafkaProperties.buildConsumerProperties(null);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  
          setSecurityProps(props);
          return new DefaultKafkaConsumerFactory<>(props);
      }
  
      @Bean
      public KafkaListenerContainerFactory<?> stringListenerFactory(ConsumerFactory<String, String> stringConsumerFactory) {
          ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
          factory.setConsumerFactory(stringConsumerFactory);
          factory.setBatchListener(false);
          return factory;
      }
  
      @Bean
      DefaultKafkaProducerFactory<String, OrderDto> orderDtoProducerFactory(KafkaProperties properties) {
          Map<String, Object> producerProperties = properties.buildProducerProperties(null);
          producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
  
  
          setSecurityProps(producerProperties);
  
          return new DefaultKafkaProducerFactory<>(producerProperties);
      }
  
      @Bean
      KafkaTemplate<String, OrderDto> orderDtoKafkaTemplate(DefaultKafkaProducerFactory<String, OrderDto> orderDtoProducerFactory) {
          return new KafkaTemplate<>(orderDtoProducerFactory);
      }
  
      @Bean
      public ConsumerFactory<String, OrderDeliveryInfoDto> orderDeliveryInfoDtoConsumerFactory(KafkaProperties kafkaProperties) {
          Map<String, Object> props = kafkaProperties.buildConsumerProperties(null);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
          props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
          props.put(JsonDeserializer.TRUSTED_PACKAGES, "ru.springio.orders.service.dto");
  
          setSecurityProps(props);
  
          return new DefaultKafkaConsumerFactory<>(props);
      }
  
      @Bean
      public KafkaListenerContainerFactory<?> orderDeliveryInfoDtoListenerFactory(ConsumerFactory<String, OrderDeliveryInfoDto> orderDeliveryInfoDtoConsumerFactory) {
          ConcurrentKafkaListenerContainerFactory<String, OrderDeliveryInfoDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
          factory.setConsumerFactory(orderDeliveryInfoDtoConsumerFactory);
          factory.setBatchListener(false);
          DefaultErrorHandler errorHandler = new DefaultErrorHandler((record, exception) -> {
              System.out.println(record);
          }, new FixedBackOff(0L, 2L));
  
          factory.setCommonErrorHandler(errorHandler);
          return factory;
      }
  
      private void setSecurityProps(Map<String, Object> props) {
          props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
          props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
  
          String jaasConfig = "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";".formatted(kafkaUsername, kafkaPassword);
          props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
      }
  }
  