[jira] [Comment Edited] (KAFKA-5124) shouldInnerLeftJoin unit test fails

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

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

Armin Braun edited comment on KAFKA-5124 at 4/25/17 8:14 PM:
-------------------------------------------------------------

@mjsax certainly looks like it at least. I only found about 30 mins for this today (haven't quite reached the root yet) but should be able to figure it out tomorrow morning after some sleep :)

What I have learnt so far is:
* I have some indication this could be a "leakish" kind of thing ... once I ran like 500 iterations to get the test to fail in this way, it fails like every 10th run, while the initial 500 runs are green. I could never make it fail without a bunch of "warmup" runs so far.
* Not a timeout issue ... test normally takes ~ 1s -> 30s are plenty -> setting Long.MAX as timeout makes it go on forever.
* This would be the thread dump with Long.MAX for the timeout (in case you can spot something my tired eyes can't at this point :D):

{code}
"main@1" prio=5 tid=0x1 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2553> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2554> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2555> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:359)
	  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:230)
	  - locked <0x2556> (a org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient)
	  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:206)
	  at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1055)
	  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1002)
	  at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.readKeyValues(IntegrationTestUtils.java:94)
	  at org.apache.kafka.streams.integration.utils.IntegrationTestUtils$1.conditionMet(IntegrationTestUtils.java:199)
	  at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:256)
	  at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(IntegrationTestUtils.java:207)
	  at org.apache.kafka.streams.integration.KTableKTableJoinIntegrationTest.verifyKTableKTableJoin(KTableKTableJoinIntegrationTest.java:220)
	  at org.apache.kafka.streams.integration.KTableKTableJoinIntegrationTest.shouldInnerLeftJoin(KTableKTableJoinIntegrationTest.java:141)
	  at sun.reflect.GeneratedMethodAccessor8.invoke(Unknown Source:-1)
	  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	  at java.lang.reflect.Method.invoke(Method.java:498)
	  at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	  at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	  at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	  at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	  at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	  at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
	  at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
	  at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
	  at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
	  at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	  at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	  at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	  at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	  at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	  at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	  at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
	  at org.junit.rules.RunRules.evaluate(RunRules.java:20)
	  at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	  at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	  at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
	  at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:71)
	  at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
	  at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)

"INNER-LEFT-ktable-ktable-join-84df535e-e89d-4e66-bf33-fc448c6b2328-StreamThread-70@9488" prio=5 tid=0x13b6 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x253b> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x253c> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x253d> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:359)
	  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:230)
	  - locked <0x253e> (a org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient)
	  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:206)
	  at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1055)
	  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1002)
	  at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:568)
	  at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:728)
	  at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:327)

"kafka-producer-network-thread | INNER-LEFT-ktable-ktable-join-84df535e-e89d-4e66-bf33-fc448c6b2328-StreamThread-70-producer@9491" daemon prio=5 tid=0x13b9 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2536> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2537> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2538> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:359)
	  at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:264)
	  at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:152)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-network-thread-0-ListenerName(PLAINTEXT)-PLAINTEXT-2@9444" prio=5 tid=0x1389 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2542> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2543> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2544> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at kafka.network.Processor.poll(SocketServer.scala:494)
	  at kafka.network.Processor.run(SocketServer.scala:432)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-network-thread-0-ListenerName(PLAINTEXT)-PLAINTEXT-1@9443" prio=5 tid=0x1388 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2545> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2546> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2547> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at kafka.network.Processor.poll(SocketServer.scala:494)
	  at kafka.network.Processor.run(SocketServer.scala:432)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-network-thread-0-ListenerName(PLAINTEXT)-PLAINTEXT-0@9442" prio=5 tid=0x1387 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2548> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2549> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x254a> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at kafka.network.Processor.poll(SocketServer.scala:494)
	  at kafka.network.Processor.run(SocketServer.scala:432)
	  at java.lang.Thread.run(Thread.java:748)

"delete-topics-thread-0@9454" prio=5 tid=0x1393 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
	  at kafka.controller.TopicDeletionManager$$anonfun$kafka$controller$TopicDeletionManager$$awaitTopicDeletionNotification$1.apply$mcV$sp(TopicDeletionManager.scala:238)
	  at kafka.controller.TopicDeletionManager$$anonfun$kafka$controller$TopicDeletionManager$$awaitTopicDeletionNotification$1.apply(TopicDeletionManager.scala:236)
	  at kafka.controller.TopicDeletionManager$$anonfun$kafka$controller$TopicDeletionManager$$awaitTopicDeletionNotification$1.apply(TopicDeletionManager.scala:236)
	  at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:213)
	  at kafka.controller.TopicDeletionManager.kafka$controller$TopicDeletionManager$$awaitTopicDeletionNotification(TopicDeletionManager.scala:235)
	  at kafka.controller.TopicDeletionManager$DeleteTopicsThread.doWork(TopicDeletionManager.scala:404)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"kafka-socket-acceptor-ListenerName(PLAINTEXT)-PLAINTEXT-0@9445" prio=5 tid=0x138a nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x253f> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2540> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2541> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at kafka.network.Acceptor.run(SocketServer.scala:274)
	  at java.lang.Thread.run(Thread.java:748)

"main-SendThread(localhost:33865)@9431" daemon prio=5 tid=0x137c nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x254b> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x254c> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x254d> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:349)
	  at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1141)

"NIOServerCxn.Factory:/127.0.0.1:0@9426" daemon prio=5 tid=0x1377 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x254e> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x254f> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2550> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.zookeeper.server.NIOServerCnxnFactory.run(NIOServerCnxnFactory.java:173)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-9@9471" daemon prio=5 tid=0x13a4 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"group-metadata-manager-0@9458" daemon prio=5 tid=0x1397 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-0@9453" daemon prio=5 tid=0x1392 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-8@9452" daemon prio=5 tid=0x1391 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-7@9451" daemon prio=5 tid=0x1390 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-6@9450" daemon prio=5 tid=0x138f nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-5@9449" daemon prio=5 tid=0x138e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-4@9439" daemon prio=5 tid=0x1386 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-3@9438" daemon prio=5 tid=0x1385 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-2@9437" daemon prio=5 tid=0x1384 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-1@9441" daemon prio=5 tid=0x1383 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-0@9436" daemon prio=5 tid=0x1382 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"SensorExpiryThread@9433" daemon prio=5 tid=0x137e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"metrics-meter-tick-thread-2@2748" daemon prio=5 tid=0x29 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"metrics-meter-tick-thread-1@2601" daemon prio=5 tid=0x1b nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"executor-Fetch@9492" prio=5 tid=0x13ba nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"ExpirationReaper-0@9456" prio=5 tid=0x1396 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9457" prio=5 tid=0x1395 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:268)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9455" prio=5 tid=0x1394 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9448" prio=5 tid=0x138d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9447" prio=5 tid=0x138c nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:268)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9446" prio=5 tid=0x138b nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"kafka-log-cleaner-thread-0@9440" prio=5 tid=0x1381 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedNanos(AbstractQueuedSynchronizer.java:1037)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(AbstractQueuedSynchronizer.java:1328)
	  at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:277)
	  at kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:260)
	  at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:220)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9379" prio=5 tid=0x1345 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9310" prio=5 tid=0x12fd nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9241" prio=5 tid=0x12b5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9172" prio=5 tid=0x126d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9103" prio=5 tid=0x1225 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9034" prio=5 tid=0x11dd nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8965" prio=5 tid=0x1195 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8896" prio=5 tid=0x114d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8827" prio=5 tid=0x1105 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8758" prio=5 tid=0x10bd nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8689" prio=5 tid=0x1075 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8620" prio=5 tid=0x102d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8551" prio=5 tid=0xfe5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8482" prio=5 tid=0xf9d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8413" prio=5 tid=0xf55 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8344" prio=5 tid=0xf0d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8275" prio=5 tid=0xec5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8206" prio=5 tid=0xe7d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8137" prio=5 tid=0xe35 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8068" prio=5 tid=0xded nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7999" prio=5 tid=0xda5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7930" prio=5 tid=0xd5d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7861" prio=5 tid=0xd15 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7792" prio=5 tid=0xccd nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7723" prio=5 tid=0xc85 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7654" prio=5 tid=0xc3d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7585" prio=5 tid=0xbf5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7516" prio=5 tid=0xbae nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7448" prio=5 tid=0xb66 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7379" prio=5 tid=0xb1e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7310" prio=5 tid=0xad6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7241" prio=5 tid=0xa8e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7172" prio=5 tid=0xa46 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7103" prio=5 tid=0x9fe nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7034" prio=5 tid=0x9b6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6965" prio=5 tid=0x96e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6896" prio=5 tid=0x926 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6827" prio=5 tid=0x8de nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6758" prio=5 tid=0x896 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6689" prio=5 tid=0x84e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6620" prio=5 tid=0x806 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6551" prio=5 tid=0x7be nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6482" prio=5 tid=0x776 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6413" prio=5 tid=0x72e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6344" prio=5 tid=0x6e6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6275" prio=5 tid=0x69e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6206" prio=5 tid=0x656 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6137" prio=5 tid=0x60e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6068" prio=5 tid=0x5c6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5998" prio=5 tid=0x57e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5930" prio=5 tid=0x536 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5861" prio=5 tid=0x4ee nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5791" prio=5 tid=0x4a6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5715" prio=5 tid=0x45e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5645" prio=5 tid=0x416 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5576" prio=5 tid=0x3ce nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5507" prio=5 tid=0x386 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5438" prio=5 tid=0x33e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5369" prio=5 tid=0x2f6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5300" prio=5 tid=0x2ae nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5231" prio=5 tid=0x266 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5159" prio=5 tid=0x21e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5090" prio=5 tid=0x1d6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5019" prio=5 tid=0x18e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@4951" prio=5 tid=0x146 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@4871" prio=5 tid=0xff nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@4801" prio=5 tid=0xb8 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@4729" prio=5 tid=0x70 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@2744" prio=5 tid=0x28 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"kafka-request-handler-7@9466" daemon prio=5 tid=0x139f nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-6@9465" daemon prio=5 tid=0x139e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-5@9464" daemon prio=5 tid=0x139d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-4@9463" daemon prio=5 tid=0x139c nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-3@9462" daemon prio=5 tid=0x139b nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-2@9461" daemon prio=5 tid=0x139a nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-1@9460" daemon prio=5 tid=0x1399 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-0@9459" daemon prio=5 tid=0x1398 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"Controller-0-to-broker-0-send-thread@9467" prio=5 tid=0x13a0 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:182)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ThrottledRequestReaper-Produce@9435" prio=5 tid=0x1380 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.server.ClientQuotaManager$ThrottledRequestReaper.doWork(ClientQuotaManager.scala:158)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ThrottledRequestReaper-Fetch@9434" prio=5 tid=0x137f nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.server.ClientQuotaManager$ThrottledRequestReaper.doWork(ClientQuotaManager.scala:158)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"main-EventThread@9432" daemon prio=5 tid=0x137d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:501)

"ZkClient-EventThread-4987-localhost:33865@9430" daemon prio=5 tid=0x137b nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at org.I0Itec.zkclient.ZkEventThread.run(ZkEventThread.java:68)

"ProcessThread(sid:0 cport:33865):@9429" prio=5 tid=0x137a nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at org.apache.zookeeper.server.PrepRequestProcessor.run(PrepRequestProcessor.java:122)

"SyncThread:0@9428" prio=5 tid=0x1379 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at org.apache.zookeeper.server.SyncRequestProcessor.run(SyncRequestProcessor.java:127)

"kafka-coordinator-heartbeat-thread | INNER-LEFT-ktable-ktable-join@9490" daemon prio=5 tid=0x13b8 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:939)

"kafka-coordinator-heartbeat-thread | ktable-ktable-consumer@9489" daemon prio=5 tid=0x13b7 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:939)

"SessionTracker@9427" prio=5 tid=0x1378 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at org.apache.zookeeper.server.SessionTrackerImpl.run(SessionTrackerImpl.java:146)

"Finalizer@9524" daemon prio=8 tid=0x3 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:143)
	  at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:164)
	  at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:209)

"Reference Handler@9525" daemon prio=10 tid=0x2 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at java.lang.Object.wait(Object.java:502)
	  at java.lang.ref.Reference.tryHandlePending(Reference.java:191)
	  at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:153)

"Signal Dispatcher@9523" daemon prio=9 tid=0x4 nid=NA runnable
  java.lang.Thread.State: RUNNABLE

{code}


was (Author: original-brownbear):
@msjax certainly looks like it at least. I only found about 30 mins for this today (haven't quite reached the root yet) but should be able to figure it out tomorrow morning after some sleep :)

What I have learnt so far is:
* I have some indication this could be a "leakish" kind of thing ... once I ran like 500 iterations to get the test to fail in this way, it fails like every 10th run, while the initial 500 runs are green. I could never make it fail without a bunch of "warmup" runs so far.
* Not a timeout issue ... test normally takes ~ 1s -> 30s are plenty -> setting Long.MAX as timeout makes it go on forever.
* This would be the thread dump with Long.MAX for the timeout (in case you can spot something my tired eyes can't at this point :D):

{code}
"main@1" prio=5 tid=0x1 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2553> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2554> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2555> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:359)
	  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:230)
	  - locked <0x2556> (a org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient)
	  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:206)
	  at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1055)
	  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1002)
	  at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.readKeyValues(IntegrationTestUtils.java:94)
	  at org.apache.kafka.streams.integration.utils.IntegrationTestUtils$1.conditionMet(IntegrationTestUtils.java:199)
	  at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:256)
	  at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(IntegrationTestUtils.java:207)
	  at org.apache.kafka.streams.integration.KTableKTableJoinIntegrationTest.verifyKTableKTableJoin(KTableKTableJoinIntegrationTest.java:220)
	  at org.apache.kafka.streams.integration.KTableKTableJoinIntegrationTest.shouldInnerLeftJoin(KTableKTableJoinIntegrationTest.java:141)
	  at sun.reflect.GeneratedMethodAccessor8.invoke(Unknown Source:-1)
	  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	  at java.lang.reflect.Method.invoke(Method.java:498)
	  at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	  at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	  at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	  at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	  at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	  at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
	  at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
	  at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
	  at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
	  at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	  at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	  at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	  at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	  at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	  at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	  at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
	  at org.junit.rules.RunRules.evaluate(RunRules.java:20)
	  at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	  at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	  at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
	  at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:71)
	  at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
	  at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)

"INNER-LEFT-ktable-ktable-join-84df535e-e89d-4e66-bf33-fc448c6b2328-StreamThread-70@9488" prio=5 tid=0x13b6 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x253b> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x253c> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x253d> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:359)
	  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:230)
	  - locked <0x253e> (a org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient)
	  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:206)
	  at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1055)
	  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1002)
	  at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:568)
	  at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:728)
	  at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:327)

"kafka-producer-network-thread | INNER-LEFT-ktable-ktable-join-84df535e-e89d-4e66-bf33-fc448c6b2328-StreamThread-70-producer@9491" daemon prio=5 tid=0x13b9 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2536> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2537> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2538> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:359)
	  at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:264)
	  at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:152)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-network-thread-0-ListenerName(PLAINTEXT)-PLAINTEXT-2@9444" prio=5 tid=0x1389 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2542> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2543> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2544> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at kafka.network.Processor.poll(SocketServer.scala:494)
	  at kafka.network.Processor.run(SocketServer.scala:432)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-network-thread-0-ListenerName(PLAINTEXT)-PLAINTEXT-1@9443" prio=5 tid=0x1388 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2545> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2546> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2547> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at kafka.network.Processor.poll(SocketServer.scala:494)
	  at kafka.network.Processor.run(SocketServer.scala:432)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-network-thread-0-ListenerName(PLAINTEXT)-PLAINTEXT-0@9442" prio=5 tid=0x1387 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x2548> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2549> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x254a> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.kafka.common.network.Selector.select(Selector.java:493)
	  at org.apache.kafka.common.network.Selector.poll(Selector.java:302)
	  at kafka.network.Processor.poll(SocketServer.scala:494)
	  at kafka.network.Processor.run(SocketServer.scala:432)
	  at java.lang.Thread.run(Thread.java:748)

"delete-topics-thread-0@9454" prio=5 tid=0x1393 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
	  at kafka.controller.TopicDeletionManager$$anonfun$kafka$controller$TopicDeletionManager$$awaitTopicDeletionNotification$1.apply$mcV$sp(TopicDeletionManager.scala:238)
	  at kafka.controller.TopicDeletionManager$$anonfun$kafka$controller$TopicDeletionManager$$awaitTopicDeletionNotification$1.apply(TopicDeletionManager.scala:236)
	  at kafka.controller.TopicDeletionManager$$anonfun$kafka$controller$TopicDeletionManager$$awaitTopicDeletionNotification$1.apply(TopicDeletionManager.scala:236)
	  at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:213)
	  at kafka.controller.TopicDeletionManager.kafka$controller$TopicDeletionManager$$awaitTopicDeletionNotification(TopicDeletionManager.scala:235)
	  at kafka.controller.TopicDeletionManager$DeleteTopicsThread.doWork(TopicDeletionManager.scala:404)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"kafka-socket-acceptor-ListenerName(PLAINTEXT)-PLAINTEXT-0@9445" prio=5 tid=0x138a nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x253f> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x2540> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2541> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at kafka.network.Acceptor.run(SocketServer.scala:274)
	  at java.lang.Thread.run(Thread.java:748)

"main-SendThread(localhost:33865)@9431" daemon prio=5 tid=0x137c nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x254b> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x254c> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x254d> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:349)
	  at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1141)

"NIOServerCxn.Factory:/127.0.0.1:0@9426" daemon prio=5 tid=0x1377 nid=NA runnable
  java.lang.Thread.State: RUNNABLE
	  at sun.nio.ch.EPollArrayWrapper.epollWait(EPollArrayWrapper.java:-1)
	  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	  - locked <0x254e> (a sun.nio.ch.EPollSelectorImpl)
	  - locked <0x254f> (a java.util.Collections$UnmodifiableSet)
	  - locked <0x2550> (a sun.nio.ch.Util$3)
	  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	  at org.apache.zookeeper.server.NIOServerCnxnFactory.run(NIOServerCnxnFactory.java:173)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-9@9471" daemon prio=5 tid=0x13a4 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"group-metadata-manager-0@9458" daemon prio=5 tid=0x1397 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-0@9453" daemon prio=5 tid=0x1392 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-8@9452" daemon prio=5 tid=0x1391 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-7@9451" daemon prio=5 tid=0x1390 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-6@9450" daemon prio=5 tid=0x138f nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-5@9449" daemon prio=5 tid=0x138e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-4@9439" daemon prio=5 tid=0x1386 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-3@9438" daemon prio=5 tid=0x1385 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-2@9437" daemon prio=5 tid=0x1384 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-1@9441" daemon prio=5 tid=0x1383 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-scheduler-0@9436" daemon prio=5 tid=0x1382 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"SensorExpiryThread@9433" daemon prio=5 tid=0x137e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"metrics-meter-tick-thread-2@2748" daemon prio=5 tid=0x29 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"metrics-meter-tick-thread-1@2601" daemon prio=5 tid=0x1b nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
	  at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"executor-Fetch@9492" prio=5 tid=0x13ba nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
	  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
	  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	  at java.lang.Thread.run(Thread.java:748)

"ExpirationReaper-0@9456" prio=5 tid=0x1396 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9457" prio=5 tid=0x1395 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:268)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9455" prio=5 tid=0x1394 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9448" prio=5 tid=0x138d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9447" prio=5 tid=0x138c nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:268)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9446" prio=5 tid=0x138b nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"kafka-log-cleaner-thread-0@9440" prio=5 tid=0x1381 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedNanos(AbstractQueuedSynchronizer.java:1037)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(AbstractQueuedSynchronizer.java:1328)
	  at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:277)
	  at kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:260)
	  at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:220)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9379" prio=5 tid=0x1345 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9310" prio=5 tid=0x12fd nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9241" prio=5 tid=0x12b5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9172" prio=5 tid=0x126d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9103" prio=5 tid=0x1225 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@9034" prio=5 tid=0x11dd nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8965" prio=5 tid=0x1195 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8896" prio=5 tid=0x114d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8827" prio=5 tid=0x1105 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8758" prio=5 tid=0x10bd nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8689" prio=5 tid=0x1075 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8620" prio=5 tid=0x102d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8551" prio=5 tid=0xfe5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8482" prio=5 tid=0xf9d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8413" prio=5 tid=0xf55 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8344" prio=5 tid=0xf0d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8275" prio=5 tid=0xec5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8206" prio=5 tid=0xe7d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8137" prio=5 tid=0xe35 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@8068" prio=5 tid=0xded nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7999" prio=5 tid=0xda5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7930" prio=5 tid=0xd5d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7861" prio=5 tid=0xd15 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7792" prio=5 tid=0xccd nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7723" prio=5 tid=0xc85 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7654" prio=5 tid=0xc3d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7585" prio=5 tid=0xbf5 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7516" prio=5 tid=0xbae nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7448" prio=5 tid=0xb66 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7379" prio=5 tid=0xb1e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7310" prio=5 tid=0xad6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7241" prio=5 tid=0xa8e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7172" prio=5 tid=0xa46 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7103" prio=5 tid=0x9fe nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@7034" prio=5 tid=0x9b6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6965" prio=5 tid=0x96e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6896" prio=5 tid=0x926 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6827" prio=5 tid=0x8de nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6758" prio=5 tid=0x896 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6689" prio=5 tid=0x84e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6620" prio=5 tid=0x806 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6551" prio=5 tid=0x7be nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6482" prio=5 tid=0x776 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6413" prio=5 tid=0x72e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6344" prio=5 tid=0x6e6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6275" prio=5 tid=0x69e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6206" prio=5 tid=0x656 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6137" prio=5 tid=0x60e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@6068" prio=5 tid=0x5c6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5998" prio=5 tid=0x57e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5930" prio=5 tid=0x536 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5861" prio=5 tid=0x4ee nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5791" prio=5 tid=0x4a6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5715" prio=5 tid=0x45e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5645" prio=5 tid=0x416 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5576" prio=5 tid=0x3ce nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5507" prio=5 tid=0x386 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5438" prio=5 tid=0x33e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5369" prio=5 tid=0x2f6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5300" prio=5 tid=0x2ae nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5231" prio=5 tid=0x266 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5159" prio=5 tid=0x21e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5090" prio=5 tid=0x1d6 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@5019" prio=5 tid=0x18e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@4951" prio=5 tid=0x146 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@4871" prio=5 tid=0xff nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@4801" prio=5 tid=0xb8 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@4729" prio=5 tid=0x70 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ExpirationReaper-0@2744" prio=5 tid=0x28 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.utils.timer.SystemTimer.advanceClock(Timer.scala:105)
	  at kafka.server.DelayedOperationPurgatory.advanceClock(DelayedOperation.scala:354)
	  at kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper.doWork(DelayedOperation.scala:378)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"kafka-request-handler-7@9466" daemon prio=5 tid=0x139f nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-6@9465" daemon prio=5 tid=0x139e nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-5@9464" daemon prio=5 tid=0x139d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-4@9463" daemon prio=5 tid=0x139c nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-3@9462" daemon prio=5 tid=0x139b nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-2@9461" daemon prio=5 tid=0x139a nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-1@9460" daemon prio=5 tid=0x1399 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"kafka-request-handler-0@9459" daemon prio=5 tid=0x1398 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:418)
	  at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:246)
	  at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:52)
	  at java.lang.Thread.run(Thread.java:748)

"Controller-0-to-broker-0-send-thread@9467" prio=5 tid=0x13a0 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:182)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ThrottledRequestReaper-Produce@9435" prio=5 tid=0x1380 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.server.ClientQuotaManager$ThrottledRequestReaper.doWork(ClientQuotaManager.scala:158)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"ThrottledRequestReaper-Fetch@9434" prio=5 tid=0x137f nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
	  at java.util.concurrent.DelayQueue.poll(DelayQueue.java:259)
	  at kafka.server.ClientQuotaManager$ThrottledRequestReaper.doWork(ClientQuotaManager.scala:158)
	  at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)

"main-EventThread@9432" daemon prio=5 tid=0x137d nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:501)

"ZkClient-EventThread-4987-localhost:33865@9430" daemon prio=5 tid=0x137b nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at org.I0Itec.zkclient.ZkEventThread.run(ZkEventThread.java:68)

"ProcessThread(sid:0 cport:33865):@9429" prio=5 tid=0x137a nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at org.apache.zookeeper.server.PrepRequestProcessor.run(PrepRequestProcessor.java:122)

"SyncThread:0@9428" prio=5 tid=0x1379 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at sun.misc.Unsafe.park(Unsafe.java:-1)
	  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.take(LinkedBlockingQueue.java:442)
	  at org.apache.zookeeper.server.SyncRequestProcessor.run(SyncRequestProcessor.java:127)

"kafka-coordinator-heartbeat-thread | INNER-LEFT-ktable-ktable-join@9490" daemon prio=5 tid=0x13b8 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:939)

"kafka-coordinator-heartbeat-thread | ktable-ktable-consumer@9489" daemon prio=5 tid=0x13b7 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:939)

"SessionTracker@9427" prio=5 tid=0x1378 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at org.apache.zookeeper.server.SessionTrackerImpl.run(SessionTrackerImpl.java:146)

"Finalizer@9524" daemon prio=8 tid=0x3 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:143)
	  at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:164)
	  at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:209)

"Reference Handler@9525" daemon prio=10 tid=0x2 nid=NA waiting
  java.lang.Thread.State: WAITING
	  at java.lang.Object.wait(Object.java:-1)
	  at java.lang.Object.wait(Object.java:502)
	  at java.lang.ref.Reference.tryHandlePending(Reference.java:191)
	  at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:153)

"Signal Dispatcher@9523" daemon prio=9 tid=0x4 nid=NA runnable
  java.lang.Thread.State: RUNNABLE

{code}

> shouldInnerLeftJoin unit test fails
> -----------------------------------
>
>                 Key: KAFKA-5124
>                 URL: https://issues.apache.org/jira/browse/KAFKA-5124
>             Project: Kafka
>          Issue Type: Sub-task
>          Components: streams
>    Affects Versions: 0.10.2.0
>            Reporter: Eno Thereska
>            Assignee: Armin Braun
>             Fix For: 0.11.0.0
>
>
> Unit test on trunk gives occasional failure:
> org.apache.kafka.streams.integration.KTableKTableJoinIntegrationTest > shouldInnerLeftJoin FAILED
>     java.lang.AssertionError: Condition not met within timeout 30000. Expecting 1 records from topic output- while only received 0: []
>         at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:265)
>         at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(IntegrationTestUtils.java:207)
>         at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(IntegrationTestUtils.java:176)
>         at org.apache.kafka.streams.integration.KTableKTableJoinIntegrationTest.verifyKTableKTableJoin(KTableKTableJoinIntegrationTest.java:222)
>         at org.apache.kafka.streams.integration.KTableKTableJoinIntegrationTest.shouldInnerLeftJoin(KTableKTableJoinIntegrationTest.java:143)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)