Deadlock when committing offsets

[Oleg Zhurakousky <oz...@hortonworks.com>]

Hi

Was wondering what could be the causes of what appears to be an intermittent deadlock when calling Consumer.commitOffsets(..)
Below are the few relevant thread dump segments:

"StandardProcessScheduler Thread-7" Id=115 BLOCKED  on java.lang.Object@2baae51
	at kafka.consumer.ZookeeperConsumerConnector.commitOffsets(ZookeeperConsumerConnector.scala:333)
	at kafka.consumer.ZookeeperConsumerConnector.commitOffsets(ZookeeperConsumerConnector.scala:324)


"ConsumerFetcherThread-74993895-50e9-3962-90e3-97af1fed7294_daves-nifi-cluster-2-1459431214410-d3b5143c-0-1001" Id=25120 WAITING  on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@23dc03a2
	at sun.misc.Unsafe.park(Native Method)
	at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
	at java.util.concurrent.LinkedBlockingQueue.put(LinkedBlockingQueue.java:350)
	at kafka.consumer.PartitionTopicInfo.enqueue(PartitionTopicInfo.scala:60)
	at kafka.consumer.ConsumerFetcherThread.processPartitionData(ConsumerFetcherThread.scala:53)
	at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1$$anonfun$apply$3.apply(AbstractFetcherThread.scala:142)
	at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1$$anonfun$apply$3.apply(AbstractFetcherThread.scala:126)
	at scala.Option.foreach(Option.scala:236)
	at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1.apply(AbstractFetcherThread.scala:126)
	at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2$$anonfun$apply$mcV$sp$1.apply(AbstractFetcherThread.scala:123)
	at scala.collection.immutable.Map$Map2.foreach(Map.scala:130)
	at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2.apply$mcV$sp(AbstractFetcherThread.scala:123)
	at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2.apply(AbstractFetcherThread.scala:123)
	at kafka.server.AbstractFetcherThread$$anonfun$processFetchRequest$2.apply(AbstractFetcherThread.scala:123)
	at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:298)
	at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:122)
	at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:98)
	at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)
	Number of Locked Synchronizers: 1
	- java.util.concurrent.locks.ReentrantLock$NonfairSync@6b99259f


"ConsumerFetcherThread-33156eec-156c-4b32-9598-d5fc3ca460ce_daves-nifi-cluster-2-1459519432175-5beddff4-0-1001" Id=35266 RUNNABLE  (in native code)
	at sun.nio.ch.Net.poll(Native Method)
	at sun.nio.ch.SocketChannelImpl.poll(SocketChannelImpl.java:954)
	- waiting on java.lang.Object@3e91ba2b
	at sun.nio.ch.SocketAdaptor$SocketInputStream.read(SocketAdaptor.java:204)
	- waiting on java.lang.Object@494070b0
	at sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:103)
	- waiting on sun.nio.ch.SocketAdaptor$SocketInputStream@3ade0aef
	at java.nio.channels.Channels$ReadableByteChannelImpl.read(Channels.java:385)
	- waiting on java.lang.Object@7d49c744
	at kafka.utils.CoreUtils$.read(CoreUtils.scala:192)
	at kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
	at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
	at kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
	at kafka.network.BlockingChannel.receive(BlockingChannel.scala:131)
	at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:81)
	at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:78)
	- waiting on java.lang.Object@4eba7a24
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:122)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:122)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:122)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:121)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:121)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:121)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
	at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:120)
	at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:106)
	at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:98)
	at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)

Thanks
Oleg