Could you check the following FAQ? https://cwiki.apache.org/confluence/display/KAFKA/FAQ#FAQ-Whydoesmyconsumernevergetanydata ? https://cwiki.apache.org/confluence/display/KAFKA/FAQ#FAQ-Myconsumerseemstohavestopped,why ?
Thanks, Jun On Mon, Jan 20, 2014 at 7:22 AM, Abhishek Bhattacharjee < abhishek.bhattacharje...@gmail.com> wrote: > Sorry I have sent both codes as consumer codes. This is the producer code. > > *Producer.java* > > package kafka.examples; > > > import java.util.Properties; > import kafka.producer.KeyedMessage; > import kafka.producer.ProducerConfig; > > public class Producer/* extends Thread*/ > { > private final kafka.javaapi.producer.Producer<Integer, String> producer; > private final String topic; > private final Properties props = new Properties(); > > public Producer(String topic) > { > props.put("serializer.class", "kafka.serializer.StringEncoder"); > props.put("metadata.broker.list", "localhost:9092"); > // Use random partitioner. Don't need the key type. Just set it to > Integer. > // The message is of type String. > producer = new kafka.javaapi.producer.Producer<Integer, String>(new > ProducerConfig(props)); > this.topic = topic; > System.out.println("Producer at "+this.topic); > } > > public void putdata() { > int messageNo = 1; > while(messageNo < 100) > { > String messageStr = new String("Message_" + messageNo); > producer.send(new KeyedMessage<Integer, String>(topic ,messageStr)); > messageNo = messageNo +1; > } > producer.close(); > System.out.println("Producer exit"); > } > > } > > > On Mon, Jan 20, 2014 at 8:46 PM, Abhishek Bhattacharjee < > abhishek.bhattacharje...@gmail.com> wrote: > > > Hello, > > I am new to kafka and facing some problem. > > My producer code works properly and sends data. > > But the consumer is not able to read it. > > Here are the codes for Producer and Consumer. > > Something is wrong with the Consumer.java code can someone please help > > with this. > > > > > > *Producer.java* > > > > package kafka.examples; > > > > > > import java.util.HashMap; > > import java.util.List; > > import java.util.Map; > > import java.util.Properties; > > import kafka.consumer.ConsumerConfig; > > import kafka.consumer.ConsumerIterator; > > import kafka.consumer.KafkaStream; > > import kafka.javaapi.consumer.ConsumerConnector; > > import kafka.message.Message; > > > > > > public class Consumer > > { > > private final ConsumerConnector consumer; > > private final String topic; > > > > public Consumer(String topic) > > { > > consumer = kafka.consumer.Consumer.createJavaConsumerConnector( > > createConsumerConfig()); > > this.topic = topic; > > System.out.println("Consumer at "+this.topic); > > } > > > > private static ConsumerConfig createConsumerConfig() > > { > > Properties props = new Properties(); > > props.put("zookeeper.connect", KafkaProperties.zkConnect); > > props.put("group.id", KafkaProperties.groupId); > > props.put("zookeeper.session.timeout.ms", "400"); > > props.put("zookeeper.sync.time.ms", "200"); > > props.put("auto.commit.interval.ms", "1000"); > > > > return new ConsumerConfig(props); > > > > } > > > > public void readdata() { > > Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); > > topicCountMap.put(topic, new Integer(1)); > > Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = > > consumer.createMessageStreams(topicCountMap); > > KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0); > > ConsumerIterator<byte[], byte[]> it = stream.iterator(); > > System.out.println("Inside read data"); > > while(it.hasNext()) > > System.out.println(new String(it.next().message())); > > > > } > > } > > > > And this is the consumer code. > > > > *Consumer.java* > > > > package kafka.examples; > > > > > > import java.util.HashMap; > > import java.util.List; > > import java.util.Map; > > import java.util.Properties; > > import kafka.consumer.ConsumerConfig; > > import kafka.consumer.ConsumerIterator; > > import kafka.consumer.KafkaStream; > > import kafka.javaapi.consumer.ConsumerConnector; > > import kafka.message.Message; > > > > > > public class Consumer > > { > > private final ConsumerConnector consumer; > > private final String topic; > > > > public Consumer(String topic) > > { > > consumer = kafka.consumer.Consumer.createJavaConsumerConnector( > > createConsumerConfig()); > > this.topic = topic; > > System.out.println("Consumer at "+topic); > > } > > > > private static ConsumerConfig createConsumerConfig() > > { > > Properties props = new Properties(); > > props.put("zookeeper.connect", KafkaProperties.zkConnect); > > props.put("group.id", KafkaProperties.groupId); > > props.put("zookeeper.session.timeout.ms", "400"); > > props.put("zookeeper.sync.time.ms", "200"); > > props.put("auto.commit.interval.ms", "1000"); > > > > return new ConsumerConfig(props); > > > > } > > > > public void readdata() { > > Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); > > topicCountMap.put(topic, new Integer(1)); > > Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = > > consumer.createMessageStreams(topicCountMap); > > KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0); > > ConsumerIterator<byte[], byte[]> it = stream.iterator(); > > while(it.hasNext()) > > System.out.println(new String(it.next().message())); > > } > > } > > > > > > Thanks. > > -- > > *Abhishek Bhattacharjee* > > *Pune Institute of Computer Technology* > > > > > > -- > *Abhishek Bhattacharjee* > *Pune Institute of Computer Technology* >