Add bare minimum Kafka producer. Downgrade Java version, since 23 doe…

…s not have security manager which is needed by Kafka. More details here - quarkusio/quarkus#39634.

.gitignore

@@ -35,3 +35,9 @@ build/
 
 ### Secure connect bundles ###
 src/main/resources/secure-connect.zip
+
+### Properties ###
+application.properties
+src/main/resources/client.truststore.jks
+src/main/resources/ca.pem
+

pom.xml

@@ -27,7 +27,7 @@
 		<url/>
 	</scm>
 	<properties>
-		<java.version>23</java.version>
+		<java.version>17</java.version>
 	</properties>
 	<dependencies>
 		<dependency>

src/main/java/com/ikea/assignment/product_information_system/controller/ProductsController.java

@@ -2,6 +2,7 @@
 
 import com.ikea.assignment.product_information_system.model.Country;
 import com.ikea.assignment.product_information_system.model.Product;
+import com.ikea.assignment.product_information_system.producers.ProductPriceUpdateProducer;
 import com.ikea.assignment.product_information_system.repository.ProductRepository;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.ResponseEntity;
@@ -16,30 +17,17 @@
 public class ProductsController {
     @Autowired
     private ProductRepository productRepository;
+    @Autowired
+    private ProductPriceUpdateProducer productPriceUpdateProducer;
 
-    private Product faltuSetup() {
-        Product product = new Product();
-        product.setSku("1");
-        product.setTitle("Title");
-        product.setPrice(BigDecimal.TEN);
-        product.setCountry(Country.EU);
-        product.setDiscountPercentage(2.0);
-        product.setImageUrl("https://assets.dummyjson.com/public/qr-code.png");
-        return product;
-    }
     @GetMapping
     public List<Product> getProducts() {
-        Product product = faltuSetup();
-        List<Product> products = new ArrayList<>();
-        products.add(product);
-        return products;
-        //return productRepository.findAll();
+        return productRepository.findAll();
     }
 
     @GetMapping(value="/{sku}")
     public Product getProduct(@PathVariable String sku) {
-        return faltuSetup();
-        //return productRepository.findById(sku).orElseThrow(RuntimeException::new);
+        return productRepository.findById(sku).orElseThrow(RuntimeException::new);
     }
 
     @PutMapping("/{sku}")
@@ -49,7 +37,9 @@ public ResponseEntity updateProduct(@PathVariable String sku, @RequestBody Produ
         currentProduct.setPrice(product.getPrice());
         currentProduct.setDiscountPercentage(product.getDiscountPercentage());
         System.out.println("This product will be saved" + currentProduct.toString());
+        //do in a transaction
         Product newProduct = productRepository.save(currentProduct);
+        productPriceUpdateProducer.send(currentProduct, newProduct);
 
         return ResponseEntity.ok(newProduct);
     }

src/main/java/com/ikea/assignment/product_information_system/model/ProductPriceUpdateEventData.java

@@ -0,0 +1,57 @@
+package com.ikea.assignment.product_information_system.model;
+
+public class ProductPriceUpdateEventData {
+    private String country;
+    private String oldPrice;
+    private String newPrice;
+    private String sku;
+    private String timestamp;
+
+    public ProductPriceUpdateEventData(String country, String oldPrice, String newPrice, String sku, String timestamp) {
+        this.country = country;
+        this.oldPrice = oldPrice;
+        this.newPrice = newPrice;
+        this.sku = sku;
+        this.timestamp = timestamp;
+    }
+
+    public String getCountry() {
+        return country;
+    }
+
+    public void setCountry(String country) {
+        this.country = country;
+    }
+
+    public String getOldPrice() {
+        return oldPrice;
+    }
+
+    public void setOldPrice(String oldPrice) {
+        this.oldPrice = oldPrice;
+    }
+
+    public String getNewPrice() {
+        return newPrice;
+    }
+
+    public void setNewPrice(String newPrice) {
+        this.newPrice = newPrice;
+    }
+
+    public String getSku() {
+        return sku;
+    }
+
+    public void setSku(String sku) {
+        this.sku = sku;
+    }
+
+    public String getTimestamp() {
+        return timestamp;
+    }
+
+    public void setTimestamp(String timestamp) {
+        this.timestamp = timestamp;
+    }
+}

src/main/java/com/ikea/assignment/product_information_system/producers/ProductPriceUpdateProducer.java

@@ -0,0 +1,27 @@
+package com.ikea.assignment.product_information_system.producers;
+
+import com.ikea.assignment.product_information_system.model.Product;
+import com.ikea.assignment.product_information_system.model.ProductPriceUpdateEventData;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+@Service
+public class ProductPriceUpdateProducer {
+    @Autowired
+    private KafkaTemplate<String, ProductPriceUpdateEventData> kafkaTemplate;
+
+    @Value("${kafka.price.update.topic.suffix}")
+    private String priceUpdateTopicSuffix;
+
+    public void send(Product oldProduct, Product newProduct) {
+        String topicName = newProduct.getCountry().toString() + "-" + priceUpdateTopicSuffix;
+        String date = new SimpleDateFormat("yyyy.MM.dd.HH.mm.ss").format(new Date());
+        ProductPriceUpdateEventData productPriceUpdateEventData = new ProductPriceUpdateEventData(newProduct.getCountry().toString(), oldProduct.getPrice().toString(), newProduct.getPrice().toString(), newProduct.getSku(), date);
+        kafkaTemplate.send(topicName, newProduct.getSku(), productPriceUpdateEventData);
+    }
+}

src/main/resources/application.checkin.properties

@@ -0,0 +1,34 @@
+spring.application.name=product-information-system
+
+spring.cassandra.keyspace-name=main
+spring.cassandra.username=<username>
+spring.cassandra.password=<password>
+spring.cassandra.schema-action=create_if_not_exists
+
+datastax.astra.secure-connect-bundle=src/main/resources/secure-connect.zip
+
+astra.db.id=<db-id>
+astra.db.region=us-east1
+astra.db.application.token=<token>
+astra.db.keyspace=main
+
+spring.kafka.bootstrap-servers=product-information-system-demo-product-information-system-demo.c.aivencloud.com:11627
+spring.kafka.properties.security.protocol=SASL_SSL
+spring.kafka.properties.sasl.mechanism=SCRAM-SHA-256
+spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=<username> password=<password>;
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonSerializer
+spring.kafka.consumer.group-id=product-price-update-group
+spring.kafka.ssl.endpoint.identification.algorithm=
+spring.kafka.ssl.truststore.type=jks
+spring.kafka.ssl.truststore.password=<password>
+spring.kafka.properties.ssl.endpoint.identification.algorithm=
+spring.kafka.properties.ssl.truststore.type=jks
+spring.kafka.properties.ssl.truststore.location=src/main/resources/client.truststore.jks
+spring.kafka.properties.ssl.truststore.password=<password>
+
+kafka.price.update.topic.suffix=product-price-updates
+kafka.country.update.topic.suffix=product-country-updates
+kafka.discount-percentage.update.topic.suffix=product-discount-updates