CustomerOrder push working KStream but Customer information is empty

Author: pavankjadda

src/main/java/com/kafkastream/events/EventsListener.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.Stores;
+import org.springframework.web.client.RestTemplate;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -89,10 +90,22 @@ public static void main(String[] args)
             System.out.println("orderKTable.value: " + order);
             CustomerOrder customerOrder = new CustomerOrder();
             customerOrder.setCustomerId(order.getCustomerId());
-            customerOrder.setFirstName("");
-            customerOrder.setLastName("");
-            customerOrder.setEmail("");
-            customerOrder.setPhone("");
+            Customer customer=getCustomerInformation(order.getCustomerId());
+            if(customer == null)
+            {
+                customerOrder.setFirstName("");
+                customerOrder.setLastName("");
+                customerOrder.setEmail("");
+                customerOrder.setPhone("");
+            }
+            else
+            {
+                customerOrder.setFirstName(customer.getFirstName());
+                customerOrder.setLastName(customer.getLastName());
+                customerOrder.setEmail(customer.getEmail());
+                customerOrder.setPhone(customer.getPhone());
+            }
+
             customerOrder.setOrderId(order.getOrderId());
             customerOrder.setOrderItemName(order.getOrderItemName());
             customerOrder.setOrderPlace(order.getOrderPlace());
@@ -135,6 +148,12 @@ public void run()
         System.exit(0);
     }
 
+    private static Customer getCustomerInformation(CharSequence customerId)
+    {
+        RestTemplate restTemplate=new RestTemplate();
+        return restTemplate.getForObject("http://localhost:8095/store/customer/"+customerId,Customer.class);
+    }
+
 
     private static <VT extends SpecificRecord> SpecificAvroSerde<VT> createSerde(String schemaRegistryUrl)
     {

src/main/java/com/kafkastream/web/kafkarest/StateStoreRestService.java

@@ -200,7 +200,10 @@ public void start() throws Exception
         jettyServer.start();
     }
 
-    public Customer getCustomerInformation(CharSequence customerId)
+    @GET
+    @Path("/customer/{customerId}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Customer getCustomerInformation(@PathParam("customerId") String customerId)
     {
         try
         {