#1149 - [Search][Recommendation] Introduce Kafka Stream to aggregate …

…all product data. -implement kafka stream application to capture all changes related to product
nashtech-garage · Oct 16, 2024 · 222fa3c · 222fa3c
1 parent c5106ab
commit 222fa3c
Show file tree

Hide file tree

Showing 6 changed files with 7 additions and 60 deletions.
diff --git a/kafka/connects/debezium-product-group.json b/kafka/connects/debezium-product-group.json
@@ -12,5 +12,6 @@
   "key.converter": "org.apache.kafka.connect.json.JsonConverter",
   "schema.include.list": "public",
   "table.include.list": "public.product_attribute, public.product_attribute_value, public.product_category, public.category, public.brand",
-  "slot.name": "product_relation_slot"
+  "slot.name": "product_relation_slot",
+  "topic.creation.default.cleanup.policy": "compact"
 }
diff --git a/kafka/connects/debezium-product.json b/kafka/connects/debezium-product.json
@@ -12,5 +12,6 @@
   "key.converter": "org.apache.kafka.connect.json.JsonConverter",
   "schema.include.list": "public",
   "table.include.list": "public.product",
-  "slot.name": "product_slot"
+  "slot.name": "product_slot",
+  "topic.creation.default.cleanup.policy": "compact"
 }
diff --git a/recommendation/src/main/java/com/yas/recommendation/dto/ProductAttributeAggregationDTO.java b/recommendation/src/main/java/com/yas/recommendation/dto/ProductAttributeAggregationDTO.java
diff --git a/recommendation/src/main/java/com/yas/recommendation/dto/ProductCategoryAggregationDTO.java b/recommendation/src/main/java/com/yas/recommendation/dto/ProductCategoryAggregationDTO.java
diff --git a/recommendation/src/main/java/com/yas/recommendation/topology/ProductTopology.java b/recommendation/src/main/java/com/yas/recommendation/topology/ProductTopology.java
@@ -110,10 +110,10 @@ protected void proccess(StreamsBuilder streamsBuilder) {
                         createMaterializedStore(PRODUCT_CATEGORY_AGGREGATION_MATERIALIZED_VIEW, Serdes.Long(), getAggregationDTOSerde(Long.class, CategoryDTO.class))
                 );
 
+
         KTable<Long, ProductResultDTO> addingCategoryDetailTable = productBrandTable
                 .leftJoin(
                         productCategoryAgg,
-                        ProductResultDTO::getId,
                         (productResult, agg) -> {
                             if (agg != null) {
                                 productResult.setCategories(agg.getAggregationContents());

diff --git a/recommendation/src/main/resources/application.properties b/recommendation/src/main/resources/application.properties
@@ -47,6 +47,8 @@ product.topic.name = dbproduct.public.product
 spring.kafka.consumer.bootstrap-servers=kafka:9092
 spring.kafka.consumer.group-id=recommendation
 
+product.sink.topic.name=test
+
 # Similarity Search Config
 yas.recommendation.embedding-based.search.topK=10
 yas.recommendation.embedding-based.search.initDefaultData=false