yuhuo created KAFKA-13428:
-----------------------------
Summary: server hang on shutdown
Key: KAFKA-13428
URL: https://issues.apache.org/jira/browse/KAFKA-13428
Project: Kafka
Issue Type: Improvement
Components: core
Affects Versions: 2.5.0
Reporter: yuhuo
Kafka Server start as follow step:
1. socketServer.startup
2. zkClient.registerBroker
3. socketServer.startDataPlaneProcessors
after step1,the port can be connnected, but default processors queue size is
20, if there is many connections，the AcceptorThread will wait on processors
queue put. and then, if registerBroker error(such as zk session not expired,
the broker id still exist), server will shutdown and never start network
processors, AcceptorThread will shutdown fail because thread still wait on
queue, at last server is hang.
stack:
{code:java}
//代码占位符
{code}
... "data-plane-kafka-socket-acceptor-ListenerName(ING_INSIDE)-PLAINTEXT-9094"
#35 prio=5 os_prio=0 tid=0x000055fe58048800 nid=0x6c5 runnable
[0x00007f5f60f8b000] java.lang.Thread.State: RUNNABLE at
sun.nio.ch.EPollArrayWrapper.epollWait(Native Method) 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
<0x00000006497b86a0> (a sun.nio.ch.Util$3) - locked <0x00000006497b8690> (a
java.util.Collections$UnmodifiableSet) - locked <0x00000006497b86b0> (a
sun.nio.ch.EPollSelectorImpl) at
sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97) at
kafka.network.Acceptor.run(SocketServer.scala:534) at
java.lang.Thread.run(Thread.java:748)
"data-plane-kafka-socket-acceptor-ListenerName(INSIDE)-PLAINTEXT-9092" #34
prio=5 os_prio=0 tid=0x000055fe55773800 nid=0x6c4 waiting on condition
[0x00007f5f63315000] java.lang.Thread.State: WAITING (parking) at
sun.misc.Unsafe.park(Native Method) - parking to wait for <0x00000006497b88d8>
(a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject) 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.ArrayBlockingQueue.put(ArrayBlockingQueue.java:353) at
kafka.network.Processor.accept(SocketServer.scala:1002) at
kafka.network.Acceptor.assignNewConnection(SocketServer.scala:633) at
kafka.network.Acceptor.$anonfun$run$1(SocketServer.scala:560) at
kafka.network.Acceptor.run(SocketServer.scala:544) at
java.lang.Thread.run(Thread.java:748) ... "main" #1 prio=5 os_prio=0
tid=0x000055fe5519a000 nid=0x69f waiting on condition [0x00007f5f8a0cf000]
java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native
Method) - parking to wait for <0x00000006497b8df8> (a
java.util.concurrent.CountDownLatch$Sync) at
java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) at
kafka.network.AbstractServerThread.shutdown(SocketServer.scala:430) at
kafka.network.Acceptor.shutdown(SocketServer.scala:517) at
kafka.network.SocketServer.$anonfun$stopProcessingRequests$2(SocketServer.scala:267)
at
kafka.network.SocketServer.$anonfun$stopProcessingRequests$2$adapted(SocketServer.scala:267)
at kafka.network.SocketServer$$Lambda$408/1620459733.apply(Unknown Source) at
scala.collection.Iterator.foreach(Iterator.scala:941) at
scala.collection.Iterator.foreach$(Iterator.scala:941) at
scala.collection.AbstractIterator.foreach(Iterator.scala:1429) at
scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala:213) at
kafka.network.SocketServer.stopProcessingRequests(SocketServer.scala:267) -
locked <0x00000006497b8e98> (a kafka.network.SocketServer) at
kafka.server.KafkaServer.$anonfun$shutdown$4(KafkaServer.scala:617) at
kafka.server.KafkaServer$$Lambda$406/1338368149.apply$mcV$sp(Unknown Source) at
kafka.utils.CoreUtils$.swallow(CoreUtils.scala:67) at
kafka.server.KafkaServer.shutdown(KafkaServer.scala:617) at
kafka.server.KafkaServer.startup(KafkaServer.scala:358) at
kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:44) at
kafka.Kafka$.main(Kafka.scala:82) at kafka.Kafka.main(Kafka.scala)
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
