added loggers

src/main/java/com/nisum/spark/consumer/Consumer.java

 package com.nisum.spark.consumer;
 
 import com.nisum.spark.dto.Product;
+
 import org.apache.spark.sql.*;
 import org.apache.spark.sql.streaming.StreamingQuery;
 import org.apache.spark.sql.types.StructType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -12,11 +15,12 @@ import java.util.concurrent.TimeoutException;
 import static com.nisum.spark.util.Constants.*;
 
 public class Consumer {
+    static Logger log = LoggerFactory.getLogger(Consumer.class);
 
     public static void main(String[] args) throws TimeoutException {
         SparkSession sparkSession = SparkSession
                 .builder().master("local[*]").appName("Spark Streaming Example 2").getOrCreate();
-
+        log.debug("spark session is created");
         Map<String, String> kafkaConfigMap = new HashMap<>();
         kafkaConfigMap.put("kafka.bootstrap.servers", HOST);
         kafkaConfigMap.put("subscribe", TOPIC);
@@ -29,7 +33,7 @@ public class Consumer {
                 .format("kafka")
                 .options(kafkaConfigMap)
                 .load();
-
+        log.debug("kafka is loaded");
         if (df != null) {
             StructType structType = Encoders.bean(Product.class).schema();
 
@@ -56,6 +60,7 @@ public class Consumer {
     }
 
     private static StreamingQuery writeStreamingData(Dataset<Product> ds) throws TimeoutException {
+        log.debug("writing to CSV file");
         return ds
                 .writeStream()
                 .outputMode("append")
@@ -66,5 +71,6 @@ public class Consumer {
                     batchDF.write().format("csv").mode("append").save("src/main/resources/productDataOutPutCSV");
                 })
                 .start();
+
     }
 }

src/main/java/com/nisum/spark/publisher/Publisher.java

 package com.nisum.spark.publisher;
 
+
 import org.apache.spark.sql.*;
 
 import org.apache.spark.storage.StorageLevel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.HashMap;
 import java.util.Map;
 import static com.nisum.spark.util.Constants.*;
 
 
 public class Publisher {
-
+    static Logger log =  LoggerFactory.getLogger(Publisher.class);
     public static void main(String[] args) {
         SparkSession sparkSession = SparkSession
                 .builder().master("local[*]").appName("Spark Streaming Example").getOrCreate();
-
+        log.debug("spark session is created");
         Dataset<Row> dataset = sparkSession.read().
                 format("csv").option("header", true).
                 load("src/main/resources/productData.csv");
@@ -22,12 +26,14 @@ public class Publisher {
     }
     private static void publishCsvRecordsToKafkaTopic(Dataset<Row> rowDataset) {
         if (null != rowDataset) {
+            log.debug("reading from CSV file");
             Dataset<Row> KafkaPublishJson = rowDataset
                     .withColumn("value", functions.to_json(functions.struct(functions.col("productId"),
                             functions.col("productName"), functions.col("productPrice"),
                             functions.col("deliveryStatus"), functions.col("date"))))
                     .alias("value").select("value")
                     .persist(StorageLevel.MEMORY_AND_DISK());
+            log.debug("publishing to kafka");
             KafkaPublishJson
                     .write().format("kafka").options(KafkaCsvConfig()).save();
         }