using partitions from config

rajadilipkolli · Sep 2, 2023 · f76309d · f76309d
1 parent 7ae9f31
commit f76309d
Show file tree

Hide file tree

Showing 3 changed files with 50 additions and 13 deletions.
diff --git a/order-service/src/main/java/com/example/orderservice/config/kafka/KafkaConfig.java b/order-service/src/main/java/com/example/orderservice/config/kafka/KafkaConfig.java
@@ -31,9 +31,11 @@ KafkaAdmin.NewTopics topics() {
                 ORDERS_TOPIC,
                 PAYMENT_ORDERS_TOPIC,
                 STOCK_ORDERS_TOPIC);
+        // streams needs topics to be created before hand, so instead of delegating to kafkaAdmin to
+        // create, manually creating
         return new KafkaAdmin.NewTopics(
-                TopicBuilder.name(ORDERS_TOPIC).partitions(3).replicas(1).build(),
-                TopicBuilder.name(PAYMENT_ORDERS_TOPIC).partitions(3).replicas(1).build(),
-                TopicBuilder.name(STOCK_ORDERS_TOPIC).partitions(3).replicas(1).build());
+                TopicBuilder.name(ORDERS_TOPIC).build(),
+                TopicBuilder.name(PAYMENT_ORDERS_TOPIC).build(),
+                TopicBuilder.name(STOCK_ORDERS_TOPIC).build());
     }
 }
diff --git a/order-service/src/main/java/com/example/orderservice/mapper/OrderMapper.java b/order-service/src/main/java/com/example/orderservice/mapper/OrderMapper.java
@@ -45,6 +45,8 @@ default void addOrderItemToOrderEntity(OrderDto orderDTO, @MappingTarget Order o
 
     @Mapping(target = "id", ignore = true)
     @Mapping(target = "items", ignore = true)
+    @Mapping(target = "status", ignore = true)
+    @Mapping(target = "source", ignore = true)
     Order orderRequestToEntity(OrderRequest orderRequest);
 
     @Mapping(target = "id", ignore = true)
@@ -61,5 +63,8 @@ default void addOrderItemRequestToOrderEntity(
                                 order.addOrderItem(orderItemRequestToOrderItem(orderItemRequest)));
     }
 
-    void updateOrderFromOrderRequest(OrderRequest orderRequest, @MappingTarget Order orderObj);
+    @Mapping(target = "status", ignore = true)
+    @Mapping(target = "source", ignore = true)
+    @Mapping(target = "id", ignore = true)
+    void updateOrderFromOrderRequest(OrderRequest orderRequest, @MappingTarget Order order);
 }
diff --git a/...ervice/src/test/java/com/example/orderservice/OrderServiceApplicationIntegrationTest.java b/...ervice/src/test/java/com/example/orderservice/OrderServiceApplicationIntegrationTest.java
@@ -14,9 +14,11 @@ Licensed under MIT License Copyright (c) 2021-2023 Raja Kolli.
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 import com.example.common.dtos.OrderDto;
+import com.example.common.dtos.OrderItemDto;
 import com.example.orderservice.common.AbstractIntegrationTest;
+import java.math.BigDecimal;
 import java.time.Duration;
-import java.util.ArrayList;
+import java.util.List;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.MethodOrderer;
 import org.junit.jupiter.api.Order;
@@ -34,7 +36,8 @@ class OrderServiceApplicationIntegrationTest extends AbstractIntegrationTest {
     @Order(1)
     void shouldFetchAllOrdersFromStream() {
         // waiting till is kafka stream is changed from PARTITIONS_ASSIGNED to RUNNING
-        await().atMost(10, SECONDS)
+        await().pollDelay(5, SECONDS)
+                .atMost(10, SECONDS)
                 .pollInterval(Duration.ofSeconds(1))
                 .untilAsserted(
                         () ->
@@ -46,28 +49,55 @@ void shouldFetchAllOrdersFromStream() {
 
     @Test
     @Order(2)
-    @Disabled
+    @Disabled("until infra for streams is set up")
     void shouldFetchAllOrdersFromStreamWhenDataIsPresent() {
 
-        // Sending events to both payment-orders and stock-orders
+        // Sending event to OrderTopic for joining
+        OrderDto orderDto = new OrderDto();
+        orderDto.setOrderId(151L);
+        orderDto.setCustomerId(1001L);
+        orderDto.setStatus("ACCEPT");
+        orderDto.setSource("PAYMENT");
+        OrderItemDto orderItemDto = new OrderItemDto();
+        orderItemDto.setItemId(1L);
+        orderItemDto.setProductId("P1");
+        orderItemDto.setProductPrice(BigDecimal.TEN);
+        orderItemDto.setQuantity(1);
+        orderDto.setItems(List.of(orderItemDto));
+
+        this.kafkaTemplate.send("orders", orderDto.getOrderId(), orderDto);
+
+        // Sending events to both payment-orders, stock-orders for streaming to process and confirm
         OrderDto paymentOrderDto = new OrderDto();
         paymentOrderDto.setOrderId(151L);
         paymentOrderDto.setCustomerId(1001L);
         paymentOrderDto.setStatus("ACCEPT");
         paymentOrderDto.setSource("PAYMENT");
-        paymentOrderDto.setItems(new ArrayList<>());
+        OrderItemDto paymentOrderItemDto = new OrderItemDto();
+        paymentOrderItemDto.setItemId(1L);
+        paymentOrderItemDto.setProductId("P1");
+        paymentOrderItemDto.setProductPrice(BigDecimal.TEN);
+        paymentOrderItemDto.setQuantity(1);
+        paymentOrderDto.setItems(List.of(paymentOrderItemDto));
+
+        this.kafkaTemplate.send("payment-orders", paymentOrderDto.getOrderId(), paymentOrderDto);
 
-        this.kafkaTemplate.send("payment-orders", paymentOrderDto);
         OrderDto stockOrderDto = new OrderDto();
         stockOrderDto.setOrderId(151L);
         stockOrderDto.setCustomerId(1001L);
         stockOrderDto.setStatus("ACCEPT");
         stockOrderDto.setSource("STOCK");
-        stockOrderDto.setItems(new ArrayList<>());
+        OrderItemDto stockOrderItemDto = new OrderItemDto();
+        stockOrderItemDto.setItemId(1L);
+        stockOrderItemDto.setProductId("P1");
+        stockOrderItemDto.setProductPrice(BigDecimal.TEN);
+        stockOrderItemDto.setQuantity(1);
+        stockOrderDto.setItems(List.of(stockOrderItemDto));
 
-        this.kafkaTemplate.send("stock-orders", stockOrderDto);
+        this.kafkaTemplate.send("stock-orders", stockOrderDto.getOrderId(), stockOrderDto);
 
-        await().atMost(60, SECONDS)
+        await().atMost(30, SECONDS)
+                .pollDelay(10, SECONDS)
                 .pollInterval(Duration.ofSeconds(5))
                 .untilAsserted(
                         () ->