[jira] [Comment Edited] (KAFKA-1343) Kafka consumer iterator thread stalls

["schandr (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-1343?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13950256#comment-13950256 ] 

schandr edited comment on KAFKA-1343 at 3/28/14 1:47 AM:
---------------------------------------------------------

Thanks for pointing me to the above link.
The  kafka.tools.ConsumerOffsetChecker tool helped me...I think I found what is causing the issue

Here is my configuration
Kafka 0.8.0
Each topic has 2 partition
one broker
0 replicas

I copied the producer code from the link below to send message to both partitions. I need this because, my consumer code, is using number of threads per partition
https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example
I started the producer and I can see the messages sent to both the partition.-- Please look the producer screen shot.

On the consumer side, i have the following code

                Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, new Integer(threadCount));
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer
				.createMessageStreams(topicCountMap);
		List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

		// now launch all the threads
		executor = Executors.newFixedThreadPool(threadCount);
		// now create an object to consume the messages
		int threadNumber = 0;
		for (final KafkaStream<byte[], byte[]> stream : streams) {
			executor.submit(new StreamIteratorTask(stream, resourceUUID, topic,consumerGroupId, threadNumber));

and the StreamIteratorTask run method looks like

	public void run() {
		ConsumerIterator<byte[], byte[]> it = stream.iterator();
        while (it.hasNext()){
        	MessageAndMetadata<byte[], byte[]> msgAndmtdata = it.next();
			byte message[] = msgAndmtdata.message();
			long offset = msgAndmtdata.offset();
			MessageAndOffset msgAndOffst = new MessageAndOffset(message, offset);
System.out.println(" Offset: " + msgAndOffset.getOffset()
				+ " Message: " + new String(msgAndOffset.getMessageBytes())
				+" for consumer group : "  + consumerGroupId);

and i am setting the following consumer config properties

private ConsumerConfig createConsumerConfig(final String consumerGroupId) {
		Properties props = new Properties();
		props.put("zookeeper.connect", 				zookeeperConfigs.get("zookeeper.connect"));
		props.put("zookeeper.session.timeout.ms", 	zookeeperConfigs.get("zookeeper.session.timeout.ms"));
		props.put("zookeeper.sync.time.ms", 		zookeeperConfigs.get("zookeeper.sync.time.ms"));
		props.put("auto.commit.interval.ms", 		zookeeperConfigs.get("auto.commit.interval.ms"));
		props.put("consumer.timeout.ms", 			zookeeperConfigs.get("consumer.timeout.ms"));
		props.put("group.id", 						consumerGroupId);
		
		return new ConsumerConfig(props);
	}
I had consumer.timeout.ms set to 3000, which was too little, once I increased it to a min(since it a dev env where I am pumping messages continously), it started working. But I noticed something else, where if there are no messages more than 1 min(I waited for 5 min), the consumers are not timing out.

Should I create a seperate bug for the time out issue? 


was (Author: schandr):
Thanks for pointing me to the above link.
The  kafka.tools.ConsumerOffsetChecker tool helped me...I think I found what is causing the issue..but trying to understand the why part though.

Here is my configuration
Kafka 0.8.0
Each topic has 2 partition
one broker
0 replicas

I copied the producer code from the link below to send message to both partitions. I need this because, my consumer code, is using number of threads per partition
https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example
I started the producer and I can see the messages sent to both the partition.

> Kafka consumer iterator thread stalls
> -------------------------------------
>
>                 Key: KAFKA-1343
>                 URL: https://issues.apache.org/jira/browse/KAFKA-1343
>             Project: Kafka
>          Issue Type: Bug
>            Reporter: schandr
>         Attachments: producerscreenshot.jpg
>
>
> My dev enviornment has Kafka 0.8.0. I am using High level consumer API. My use case has one topic with 2 partitions. My code is using two threads to read from the kafka stream. The runnable iterator task has the below implementation
> 	public void run() {
> 		ConsumerIterator<byte[], byte[]> it = stream.iterator();
>         while (it.hasNext()){
>         	MessageAndMetadata<byte[], byte[]> msgAndmtdata = it.next();
> 			byte message[] = msgAndmtdata.message();
> 			long offset = msgAndmtdata.offset();
> 			MessageAndOffset msgAndOffst = new MessageAndOffset(message, offset);
> 			System.out.println(" Message offset for topic : " + topicName + ": " + offset + "thread :" + thread + "Message: " + new String(message));
> 			
>         }
> 	}
> I am using kafka console producer to produce messages. When the test starts, the above logs the messages produced so far, but stalls for new messages produced by the console producer. No exception in the kafka logs.



--
This message was sent by Atlassian JIRA
(v6.2#6252)