producer / consumer kafka reactor configuration

src/main/java/com/ascendfinalproject/warehouse/controllers/KafkaController.java

 package com.ascendfinalproject.warehouse.controllers;
 
 
-import com.ascendfinalproject.warehouse.Producer;
+import com.ascendfinalproject.warehouse.services.Producer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestMapping;

src/main/java/com/ascendfinalproject/warehouse/Consumer.java → src/main/java/com/ascendfinalproject/warehouse/services/Consumer.java

-package com.ascendfinalproject.warehouse;
+package com.ascendfinalproject.warehouse.services;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

src/main/java/com/ascendfinalproject/warehouse/services/KafkaConsumerConfig.java

+package com.ascendfinalproject.warehouse.services;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaConsumerConfig {
+
+  private Map<String, Object> consumerConfig() {
+    Map<String, Object> config = new HashMap<>();
+    config.put(ConsumerConfig.GROUP_ID_CONFIG, "WAREHOUSE_MANAGEMENT");
+    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+    config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+    return config;
+  }
+
+}

src/main/java/com/ascendfinalproject/warehouse/services/KafkaProducerConfig.java

+package com.ascendfinalproject.warehouse.services;
+
+import com.fasterxml.jackson.databind.JsonSerializer;
+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.core.KafkaTemplate;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfig {
+
+
+  private Map<String, Object> producerConfig() {
+    Map<String, Object> config = new HashMap<>();
+    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+    return config;
+  }
+
+}

src/main/java/com/ascendfinalproject/warehouse/Producer.java → src/main/java/com/ascendfinalproject/warehouse/services/Producer.java

-package com.ascendfinalproject.warehouse;
+package com.ascendfinalproject.warehouse.services;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

src/main/resources/application.properties

 spring.data.mongodb.uri=mongodb+srv://warehouse1:ascendWarehouseProject@warehouse-cluster.xopll.mongodb.net/myFirstDatabase?retryWrites=true&w=majority
 spring.data.mongodb.database=test
 
-server:
-  port: 9000
-spring:
-    kafka:
-        consumer:
-            bootstrap-servers: localhost:9092
-            group-id: WAREHOUSE_MANAGEMENT
-            auto-offset-reset: earliest
-            key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
-            value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
-        producer:
-            bootstrap-servers: localhost:9092
-            key-serializer: org.apache.kafka.common.serialization.StringSerializer
-            value-serializer: org.apache.kafka.common.serialization.StringSerializer
+#server:
+#  port: 9000
+#spring:
+#    kafka:
+#        consumer:
+#            bootstrap-servers: localhost:9092
+#            group-id: WAREHOUSE_MANAGEMENT
+#            auto-offset-reset: earliest
+#            key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+#            value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+#        producer:
+#            bootstrap-servers: localhost:9092
+#            key-serializer: org.apache.kafka.common.serialization.StringSerializer
+#            value-serializer: org.apache.kafka.common.serialization.StringSerializer
\ No newline at end of file