content: package org.springframework.samples.petclinic.system;
  
  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.boot.autoconfigure.kafka.KafkaProperties;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Configuration;
  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.support.serializer.JsonDeserializer;
  import org.springframework.kafka.support.serializer.JsonSerializer;
  import org.springframework.samples.petclinic.owner.VisitDto;
  
  import java.util.Map;
  
  @Configuration
  public class KafkaConfiguration {
  
  	@Bean
  	DefaultKafkaProducerFactory<String, String> stringProducerFactory(KafkaProperties properties) {
  		Map<String, Object> producerProperties = properties.buildProducerProperties();
  		producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  		producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  		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();
  		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  		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, VisitDto> visitDtoProducerFactory(KafkaProperties properties) {
  		Map<String, Object> producerProperties = properties.buildProducerProperties();
  		producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  		producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
  		return new DefaultKafkaProducerFactory<>(producerProperties);
  	}
  
  	@Bean
  	KafkaTemplate<String, VisitDto> visitDtoKafkaTemplate(DefaultKafkaProducerFactory<String, VisitDto> visitDtoProducerFactory) {
  		return new KafkaTemplate<>(visitDtoProducerFactory);
  	}
  
      @Bean
      public ConsumerFactory<String, VisitDto> visitDtoConsumerFactory(KafkaProperties kafkaProperties) {
          Map<String, Object> props = kafkaProperties.buildConsumerProperties();
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
          props.put(JsonDeserializer.TRUSTED_PACKAGES, "org.springframework.samples.petclinic.owner");
          return new DefaultKafkaConsumerFactory<>(props);
      }
  
      @Bean
      public KafkaListenerContainerFactory<?> visitDtoListenerFactory(ConsumerFactory<String, VisitDto> visitDtoConsumerFactory) {
          ConcurrentKafkaListenerContainerFactory<String, VisitDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
          factory.setConsumerFactory(visitDtoConsumerFactory);
          factory.setBatchListener(false);
          return factory;
      }
  }
  