Initial Commit

Author: LearningJournal

AvroConsumer-V2/ClickRecord.java

@@ -0,0 +1,40 @@
+import java.util.*;
+import org.apache.kafka.clients.consumer.*;
+
+
+public class ClickRecordConsumerV2{    
+    
+    public static void main(String[] args) throws Exception{
+
+        String topicName = "AvroClicks";
+            
+        String groupName = "RG";
+        Properties props = new Properties();
+        props.put("bootstrap.servers", "localhost:9092,localhost:9093");
+        props.put("group.id", groupName);
+        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
+        props.put("schema.registry.url", "http://localhost:8081");
+        props.put("specific.avro.reader", "true");
+        
+        KafkaConsumer<String, ClickRecord> consumer = new KafkaConsumer<>(props);
+        consumer.subscribe(Arrays.asList(topicName));
+        try{
+            while (true){
+                ConsumerRecords<String, ClickRecord> records = consumer.poll(100);
+                for (ConsumerRecord<String, ClickRecord> record : records){
+                        System.out.println("Session id="+ record.value().getSessionId()
+                                         + " Channel=" + record.value().getChannel() 
+                                         + " Entry URL=" + record.value().getEntryUrl()
+                                         + " Language=" + record.value().getLanguage());
+                    }
+                }
+            }catch(Exception ex){
+                ex.printStackTrace();
+            }
+            finally{
+                consumer.close();
+            }
+    }
+    
+}

AvroConsumer-V2/ClickRecordConsumerV2.java

@@ -0,0 +1,13 @@
+{"type": "record",
+ "name": "ClickRecord",
+ "fields": [
+     {"name": "session_id", "type": "string"},
+     {"name": "browser", "type": ["string", "null"]},
+     {"name": "campaign", "type": ["string", "null"]},
+     {"name": "channel", "type": "string"},
+     {"name": "entry_url", "type": ["string", "null"], "default": "None"},
+     {"name": "ip", "type": ["string", "null"]},
+     {"name": "language", "type": ["string", "null"], "default": "None"},
+     {"name": "os", "type": ["string", "null"],"default": "None"}     
+ ]
+}

AvroConsumer-V2/ClickRecordV2.avsc

@@ -0,0 +1,18 @@
+name := "AvroTest"
+
+val repositories = Seq(
+  "confluent" at "http://packages.confluent.io/maven/",
+  Resolver.sonatypeRepo("public")
+)
+
+libraryDependencies ++= Seq(
+"org.apache.avro" % "avro" % "1.8.1",
+"io.confluent" % "kafka-avro-serializer" % "3.1.1",
+"org.apache.kafka" % "kafka-clients" % "0.10.1.0"
+    exclude("javax.jms", "jms")
+    exclude("com.sun.jdmk", "jmxtools")
+    exclude("com.sun.jmx", "jmxri")
+    exclude("org.slf4j", "slf4j-simple")
+)
+
+resolvers += "confluent" at "http://packages.confluent.io/maven/"