package com.mj;


import org.apache.kafka.clients.producer.KafkaProducer ;
// import kafka.producer.KeyedMessage;
import org.apache.kafka.clients.producer.ProducerRecord;


import java.util.Properties;

public class KafkaProducer082 {

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		
		Properties props = new Properties();
		 
		// props.put("metadata.broker.list", "host2:9092,host3:9092");
    props.put("bootstrap.servers", "localhost:9092");
		// props.put("serializer.class", "kafka.serializer.StringEncoder");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		// props.put("partitioner.class", "example.producer.SimplePartitioner");
		props.put("request.required.acks", "1");

		
		KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
		
		String date = "04092014" ;
		// String topic = "my-replicated-topic" ;
		String topic = "mjtopic" ;
		
		
		for (int i = 1 ; i <= 1000000 ; i++) {
			
			String msg = date + " This is message " + i ;
			System.out.println(msg) ;
			
			ProducerRecord<String, String> data = new ProducerRecord<String, String>(topic, String.valueOf(i), msg);
			 
			producer.send(data);
			
			
		}
		
		producer.close();
		

	}

}

@@ -7,24 +7,75 @@
 e com.mj;%0A%0A%0A
+import org.apache.kafka.clients.producer.Callback;%0A
 import org.a
@@ -219,37 +219,130 @@
 rd;%0A
-%0A%0Aimport java.util.Properties
+import org.apache.kafka.clients.producer.RecordMetadata;%0A%0A%0Aimport java.util.Properties;%0Aimport java.util.concurrent.Future
 ;%0A%0Ap
@@ -1230,23 +1230,23 @@
  i++) %7B%0A
-%09%09%09%0A%09%09%09
+%0A      
 String m
@@ -1285,42 +1285,11 @@
 + i 
+ 
 ;%0A
-%09%09%09System.out.println(msg) ;%0A%09%09%09
 %0A%09%09%09
@@ -1405,36 +1405,567 @@
 %0A%09%09%09
-producer.send(data);%0A%09%09%09%0A%09%09%09
+Future%3CRecordMetadata%3E rs = producer.send(data, new Callback() %7B%0A        @Override%0A        public void onCompletion(RecordMetadata recordMetadata, Exception e) %7B%0A%0A          System.out.println(%22Received ack for partition=%22 + recordMetadata.partition() + %22 offset = %22 + recordMetadata.offset()) ;%0A        %7D%0A      %7D);%0A%0A%0A      try %7B%0A        RecordMetadata rm = rs.get();%0A%0A        msg = msg + %22  partition = %22 + rm.partition() +  %22 offset =%22 + rm.offset() ;%0A        System.out.println(msg) ;%0A%0A%0A      %7D catch(Exception e) %7B%0A        System.out.println(e) ;%0A      %7D%0A%0A
 %0A%09%09%7D

update 082 producer