feat: Added kafka configs to Payment Service

payment/build.gradle

@@ -8,4 +8,8 @@ dependencies {
 
 	testImplementation 'org.springframework.boot:spring-boot-starter-test'
 	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
+
+	// kafka 설정 추가
+	implementation 'org.springframework.kafka:spring-kafka'
+	testImplementation 'org.springframework.kafka:spring-kafka'
 }

payment/src/main/java/com/f_lab/la_planete/payment/config/KafkaConsumerConfig.java

@@ -0,0 +1,57 @@
+package com.f_lab.la_planete.payment.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.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.support.serializer.JsonDeserializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaConsumerConfig {
+
+  @Value("${spring.kafka.consumer.bootstrap-servers}")
+  private String BOOTSTRAP_SERVERS;
+
+  @Value("${spring.kafka.consumer.enable-auto-commit}")
+  private boolean AUTO_COMMIT;
+
+  @Value("${kafka.container.concurrency:3}")
+  private int CONCURRENCY;
+
+  @Bean
+  public Map<String, Object> consumerConfig() {
+    Map<String, Object> consumerConfig = new HashMap<>();
+
+    consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
+    consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+    consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+    consumerConfig.put(JsonDeserializer.TRUSTED_PACKAGES, "com.f_lab.la_planete.core.*");
+    consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, AUTO_COMMIT);
+
+    return consumerConfig;
+  }
+
+  @Bean
+  public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
+    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
+        new ConcurrentKafkaListenerContainerFactory<>();
+
+    factory.setConsumerFactory(consumerFactory());
+    factory.setConcurrency(CONCURRENCY);
+
+    return factory;
+  }
+
+  @Bean
+  public ConsumerFactory<String, Object> consumerFactory() {
+    return new DefaultKafkaConsumerFactory<>(consumerConfig());
+  }
+}

payment/src/main/java/com/f_lab/la_planete/payment/config/KafkaProducerConfig.java

@@ -0,0 +1,78 @@
+package com.f_lab.la_planete.payment.config;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.ProducerConfig;
+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.TopicBuilder;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfig {
+
+
+  @Value("${spring.kafka.producer.bootstrap-servers:localhost:9092}")
+  private String BOOTSTRAP_SERVERS;
+
+  @Value("${spring.kafka.producer.ack:all}")
+  private String ACK;
+
+  @Value("${spring.kafka.producer.enable.idempotence:true}")
+  private String IDEMPOTENCE;
+
+  @Value("${payments.commands.topic.name}")
+  private String paymentProcessCommand;
+
+  @Value("${payments.events.topic.name}")
+  private String paymentProcessedEvents;
+
+  @Value("${kafka.topic.partitions:3}")
+  private int TOPIC_PARTITIONS;
+
+
+  @Bean
+  public Map<String, Object> producerConfig() {
+    Map<String, Object> config = new HashMap<>();
+
+    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
+    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, IDEMPOTENCE);
+    config.put(ProducerConfig.ACKS_CONFIG, ACK);
+
+    return config;
+  }
+
+  public ProducerFactory<String, Object> producerFactory() {
+    return new DefaultKafkaProducerFactory<>(producerConfig());
+  }
+
+  @Bean
+  public KafkaTemplate<String, Object> kafkaTemplate() {
+    return new KafkaTemplate<>(producerFactory());
+  }
+
+  @Bean
+  public NewTopic foodsProcessFoodsReserveCommand() {
+    return TopicBuilder
+        .name(paymentProcessCommand)
+        .partitions(TOPIC_PARTITIONS)
+        .build();
+  }
+
+  @Bean
+  public NewTopic foodsFoodsReservationProcessedEvent() {
+    return TopicBuilder
+        .name(paymentProcessedEvents)
+        .partitions(TOPIC_PARTITIONS)
+        .build();
+  }
+}

payment/src/main/resources/application.yml

@@ -0,0 +1,37 @@
+server.port: 0
+
+spring:
+  application:
+    name: payments
+
+  kafka:
+    producer:
+      bootstrap-servers: localhost:9092
+      acks: all
+      enable:
+        idempotence: true
+
+    consumer:
+      bootstrap-servers: localhost:9092
+      group-id: payments
+      enable-auto-commit: false
+      properties:
+        spring.json.trusted.packages: com.f-lab.la_planete.core.*
+
+payments:
+  commands:
+    topic:
+      name: payments.payment-process-command
+
+  events:
+    topic:
+      name: payments.payment-processed-event
+
+
+
+kafka:
+  topic:
+    partitions: 3
+
+  container:
+    concurrency: 3