Fixed all bugs

Author: pavankjadda

src/main/java/com/kafkastream/constants/KafkaConstants.java

@@ -5,6 +5,7 @@ public class KafkaConstants
     //Kafka Streams Configuration
     public static String SCHEMA_REGISTRY_URL = "http://localhost:8081";
     public static String APPLICATION_ID_CONFIG = "cqrs-streams";
+    public static String APPLICATION_ID_CONFIG2 = "cqrs-streams2";
     public static String APPLICATION_SERVER_CONFIG = "localhost:8095";
     public static String BOOTSTRAP_SERVERS_CONFIG = "localhost:9092";
     public static String COMMIT_INTERVAL_MS_CONFIG = "2000";

src/main/java/com/kafkastream/events/EventsListener.java renamed to src/main/java/com/kafkastream/events/services/EventsListener.java

@@ -1,6 +1,7 @@
-package com.kafkastream.events;
+package com.kafkastream.events.services;
 
 import com.kafkastream.constants.KafkaConstants;
+import com.kafkastream.dto.CustomerDto;
 import com.kafkastream.model.Customer;
 import com.kafkastream.model.CustomerOrder;
 import com.kafkastream.model.Greetings;
@@ -24,6 +25,8 @@
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.Stores;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.client.RestTemplate;
 
 import java.util.Collections;
@@ -44,7 +47,7 @@ public class EventsListener
     private static void setUp()
     {
         properties = new Properties();
-        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, KafkaConstants.APPLICATION_ID_CONFIG);
+        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, KafkaConstants.APPLICATION_ID_CONFIG2);
         properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.BOOTSTRAP_SERVERS_CONFIG);
         properties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, KafkaConstants.APPLICATION_SERVER_CONFIG);
         properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, KafkaConstants.COMMIT_INTERVAL_MS_CONFIG);
@@ -59,9 +62,12 @@ private static void setUp()
 
     public static void main(String[] args)
     {
+        //StateStoreService stateStoreService=new StateStoreService();
+
         //Setup StreamsBuilder
         setUp();
 
+
         SpecificAvroSerde<Customer> customerSerde = createSerde(KafkaConstants.SCHEMA_REGISTRY_URL);
         SpecificAvroSerde<Order> orderSerde = createSerde(KafkaConstants.SCHEMA_REGISTRY_URL);
         SpecificAvroSerde<Greetings> greetingsSerde = createSerde(KafkaConstants.SCHEMA_REGISTRY_URL);
@@ -90,7 +96,7 @@ public static void main(String[] args)
             System.out.println("orderKTable.value: " + order);
             CustomerOrder customerOrder = new CustomerOrder();
             customerOrder.setCustomerId(order.getCustomerId());
-            Customer customer=getCustomerInformation(order.getCustomerId());
+            CustomerDto customer=getCustomerInformation(order.getCustomerId());
             if(customer == null)
             {
                 customerOrder.setFirstName("");
@@ -134,7 +140,6 @@ public static void main(String[] args)
             e.printStackTrace();
         }
 
-
         //Close Runtime
         Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook")
         {
@@ -148,10 +153,16 @@ public void run()
         System.exit(0);
     }
 
-    private static Customer getCustomerInformation(CharSequence customerId)
+    private static CustomerDto getCustomerInformation(CharSequence customerId)
     {
         RestTemplate restTemplate=new RestTemplate();
-        return restTemplate.getForObject("http://localhost:8095/store/customer/"+customerId,Customer.class);
+        if(customerId!=null)
+        {
+            ResponseEntity<CustomerDto> customerResponseEntity= restTemplate.exchange("http://" +KafkaConstants.REST_PROXY_HOST+":"+KafkaConstants.REST_PROXY_PORT+ "store/customer/"+customerId, HttpMethod.GET,null, CustomerDto.class);
+            return customerResponseEntity.getBody();
+
+        }
+        return null;
     }
 
 

src/main/java/com/kafkastream/events/EventsSender.java renamed to src/main/java/com/kafkastream/events/services/EventsSender.java

@@ -1,4 +1,4 @@
-package com.kafkastream.events;
+package com.kafkastream.events.services;
 
 import com.kafkastream.constants.KafkaConstants;
 import com.kafkastream.model.Customer;

src/main/java/com/kafkastream/web/EventsController.java

@@ -1,6 +1,6 @@
 package com.kafkastream.web;
 
-import com.kafkastream.events.EventsSender;
+import com.kafkastream.events.services.EventsSender;
 import com.kafkastream.model.Customer;
 import com.kafkastream.model.Greetings;
 import com.kafkastream.model.Order;

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

@@ -70,9 +70,9 @@ private static <T> T waitUntilStoreIsQueryable(final String storeName, final Que
     @GET
     @Path("/customer-order/{customerId}")
     @Produces(MediaType.APPLICATION_JSON)
-    public List<CustomerOrderDTO> getCustomerOrders(@PathParam("customerId") String customerId) throws InterruptedException
+    public List<CustomerOrderDTO> getCustomerOrder(@PathParam("customerId") String customerId) throws InterruptedException
     {
-        System.out.println("Inside getCustomerOrders()");
+        System.out.println("Inside getCustomerOrder()");
         List<CustomerOrderDTO> customerOrderList = new ArrayList<>();
         ReadOnlyKeyValueStore<String, CustomerOrder> customerOrdersStore = waitUntilStoreIsQueryable(KafkaConstants.CUSTOMER_ORDER_STORE_NAME, QueryableStoreTypes.keyValueStore(), streams);
 
@@ -203,7 +203,29 @@ public void start() throws Exception
     @GET
     @Path("/customer/{customerId}")
     @Produces(MediaType.APPLICATION_JSON)
-    public Customer getCustomerInformation(@PathParam("customerId") String customerId)
+    public CustomerDto getCustomerInformation(@PathParam("customerId") String customerId)
+    {
+        try
+        {
+            ReadOnlyKeyValueStore<String, Customer> customersStore = waitUntilStoreIsQueryable(KafkaConstants.CUSTOMER_STORE_NAME, QueryableStoreTypes.keyValueStore(), streams);
+            KeyValueIterator<String, Customer> keyValueIterator = customersStore.all();
+            while (keyValueIterator.hasNext())
+            {
+                Customer customer=keyValueIterator.next().value;
+                if(customer.getCustomerId().toString().equals(customerId))
+                    return new CustomerDto(customer.getCustomerId().toString(),customer.getFirstName().toString(),customer.getLastName().toString(),
+                            customer.getEmail().toString(),customer.getPhone().toString());
+            }
+        }
+        catch (Exception e)
+        {
+            e.printStackTrace();
+        }
+        return null;
+    }
+
+
+    public Customer getCustomer(String customerId)
     {
         try
         {

src/test/java/com/kafkastream/TestProducer.java

@@ -1,6 +1,6 @@
 package com.kafkastream;
 
-import com.kafkastream.events.EventsSender;
+import com.kafkastream.events.services.EventsSender;
 import com.kafkastream.model.Customer;
 import com.kafkastream.model.Order;
 import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;