Initial Commit

Author: LearningJournal

ConsumerExample/RandomConsumer.java

@@ -0,0 +1,44 @@
+import java.util.*;
+import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.common.*;
+
+public class RandomConsumer{
+    
+    
+    public static void main(String[] args) throws Exception{
+
+            String topicName = "RandomProducerTopic";
+            KafkaConsumer<String, String> consumer = null;
+            
+            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", "org.apache.kafka.common.serialization.StringDeserializer");
+            props.put("enable.auto.commit", "false");
+
+            consumer = new KafkaConsumer<>(props);
+            RebalanceListner rebalanceListner = new RebalanceListner(consumer);
+            
+            consumer.subscribe(Arrays.asList(topicName),rebalanceListner);
+            try{
+                while (true){
+                    ConsumerRecords<String, String> records = consumer.poll(100);
+                    for (ConsumerRecord<String, String> record : records){
+                        //System.out.println("Topic:"+ record.topic() +" Partition:" + record.partition() + " Offset:" + record.offset() + " Value:"+ record.value());
+                       // Do some processing and save it to Database
+                        rebalanceListner.addOffset(record.topic(), record.partition(),record.offset());
+                    }
+                        //consumer.commitSync(rebalanceListner.getCurrentOffsets());
+                }
+            }catch(Exception ex){
+                System.out.println("Exception.");
+                ex.printStackTrace();
+            }
+            finally{
+                    consumer.close();
+            }
+    }
+    
+}

ConsumerExample/RandomProducer.java

@@ -0,0 +1,39 @@
+import java.util.*;
+import org.apache.kafka.clients.producer.*;
+public class RandomProducer {
+  
+   public static void main(String[] args) throws InterruptedException{
+           
+      String topicName = "RandomProducerTopic";
+      String msg;
+      
+      Properties props = new Properties();
+      props.put("bootstrap.servers", "localhost:9092,localhost:9093");
+      props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");         
+      props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+            
+      Producer<String, String> producer = new KafkaProducer <>(props);
+      Random rg = new Random();
+      Calendar dt = Calendar.getInstance();
+      dt.set(2016,1,1);
+      try{
+          while(true){
+              for (int i=0;i<100;i++){
+                msg = dt.get(Calendar.YEAR)+"-"+dt.get(Calendar.MONTH)+"-"+dt.get(Calendar.DATE) + "," + rg.nextInt(1000);
+                producer.send(new ProducerRecord<String, String>(topicName,0,"Key",msg)).get();
+                msg = dt.get(Calendar.YEAR)+"-"+dt.get(Calendar.MONTH)+"-"+dt.get(Calendar.DATE) + "," + rg.nextInt(1000);
+                producer.send(new ProducerRecord<String, String>(topicName,1,"Key",msg)).get();
+              }
+              dt.add(Calendar.DATE,1);
+              System.out.println("Data Sent for " + dt.get(Calendar.YEAR) + "-" + dt.get(Calendar.MONTH) + "-" + dt.get(Calendar.DATE) );
+          }
+      }
+      catch(Exception ex){
+          System.out.println("Intrupted");
+      }
+      finally{
+          producer.close();
+        }
+      
+   }
+}

ConsumerExample/RebalanceListner.java

@@ -0,0 +1,40 @@
+import java.util.*;
+import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.common.*;
+
+public class RebalanceListner implements ConsumerRebalanceListener {
+    private KafkaConsumer consumer;
+    private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap();
+
+    public RebalanceListner(KafkaConsumer con){
+        this.consumer=con;
+    }
+    
+    public void addOffset(String topic, int partition, long offset){
+        currentOffsets.put(new TopicPartition(topic, partition),new OffsetAndMetadata(offset,"Commit"));
+    }
+    
+    public Map<TopicPartition, OffsetAndMetadata> getCurrentOffsets(){
+        return currentOffsets;
+    }
+    
+    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+        System.out.println("Following Partitions Assigned ....");
+        for(TopicPartition partition: partitions)                
+            System.out.println(partition.partition()+",");
+    }
+
+    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+        System.out.println("Following Partitions Revoked ....");
+        for(TopicPartition partition: partitions)                
+            System.out.println(partition.partition()+",");
+                
+        
+        System.out.println("Following Partitions commited ...." );
+        for(TopicPartition tp: currentOffsets.keySet())
+            System.out.println(tp.partition());
+        
+        consumer.commitSync(currentOffsets);
+        currentOffsets.clear();
+    }
+}