A popular case is that the thread isn't catching thrown Exceptions in run() - which causes the thread to be restarted in a typical Executor configuration - and does not pull any more items.
Is that the case? On Thu, Nov 29, 2012 at 11:48 AM, Juan Valencia <[email protected]>wrote: > ---------- Forwarded message ---------- > From: Juan Valencia <[email protected]> > Date: Thu, Nov 29, 2012 at 11:45 AM > Subject: Consumer Hanging (yes, read the FAQ) > To: [email protected] > > > > My consumers randomly hang. They essentially end up waiting on a kafka > stream: > There is plenty of data on the brokers and a ratio of about 1 consumer for > 5 partitions. > Any thoughts on how to keep the stream going? Or at least restart on hang? > e.g. > > ConsumerConfig consumerConfig = new ConsumerConfig(props); > ConsumerConnector consumerConnector = Consumer > .createJavaConsumerConnector(consumerConfig); > Map<String, List<KafkaStream<Message>>> topicMessageStreams = > consumerConnector > .createMessageStreams(ImmutableMap.of(topic, 10)); > List<KafkaStream<Message>> streams = topicMessageStreams.get(topic); > ExecutorService executor = Executors.newFixedThreadPool(10); > > for (final KafkaStream<Message> stream : streams) { > executor.submit(new Runnable() { > @Override > public void run() { > try { > ****** for (MessageAndMetadata<Message> msgAndMetadata : stream) { > try { .... > > The offset checker shows that the offsets stop moving, but the partitions > still have owners. > > A jstack trace shows the following (*'s below correspond to *'s above): > > "pool-1-thread-10" prio=10 tid=0x00007effa4203000 nid=0x50f5 waiting on > condition [0x00007eff9965b000] > java.lang.Thread.State: WAITING (parking) > at sun.misc.Unsafe.park(Native Method) > - parking to wait for <0x000000008b822548> (a > java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject) > at > java.util.concurrent.locks.LockSupport.park(LockSupport.java:186) > at > > java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043) > at > java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:386) > at > kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:60) > at > kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:32) > at > kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:59) > at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:51) > ****** at > com.mydomain.kafka.consumer.myConsumer$1.run(myConsumer.java:101) > at > java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) > at > java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:334) > at java.util.concurrent.FutureTask.run(FutureTask.java:166) > at > > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110) > at > > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603) > at java.lang.Thread.run(Thread.java:679) > > > > > > > > > -- > > Learn More: SQI (Social Quality Index) - A Universal Measure of Social > Quality >
