You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ne...@apache.org on 2014/03/26 23:37:47 UTC
git commit: KAFKA-1317 KafkaServer 0.8.1 not responding to
.shutdown() cleanly, possibly related to TopicDeletionManager or MetricsMeter
state; reviewed by Neha Narkhede
Repository: kafka
Updated Branches:
refs/heads/0.8.1 655e1a8aa -> 39a560789
KAFKA-1317 KafkaServer 0.8.1 not responding to .shutdown() cleanly, possibly related to TopicDeletionManager or MetricsMeter state; reviewed by Neha Narkhede
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/39a56078
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/39a56078
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/39a56078
Branch: refs/heads/0.8.1
Commit: 39a560789e65e28cc91468933487c0b23e1a1db0
Parents: 655e1a8
Author: Timothy Chen <tn...@gmail.com>
Authored: Wed Mar 26 15:35:55 2014 -0700
Committer: Neha Narkhede <ne...@gmail.com>
Committed: Wed Mar 26 15:36:05 2014 -0700
----------------------------------------------------------------------
.../kafka/controller/KafkaController.scala | 19 ++++++-----------
.../kafka/controller/TopicDeletionManager.scala | 22 +++++++++++++-------
.../unit/kafka/server/ServerShutdownTest.scala | 20 ++++++++++++++++++
3 files changed, 41 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/39a56078/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index f12ffc2..2867ef1 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -35,9 +35,9 @@ import org.I0Itec.zkclient.{IZkDataListener, IZkStateListener, ZkClient}
import org.I0Itec.zkclient.exception.{ZkNodeExistsException, ZkNoNodeException}
import java.util.concurrent.atomic.AtomicInteger
import org.apache.log4j.Logger
+import java.util.concurrent.locks.ReentrantLock
import scala.Some
import kafka.common.TopicAndPartition
-import java.util.concurrent.locks.ReentrantLock
class ControllerContext(val zkClient: ZkClient,
val zkSessionTimeout: Int) {
@@ -335,15 +335,16 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
*/
def onControllerResignation() {
inLock(controllerContext.controllerLock) {
- if (config.autoLeaderRebalanceEnable)
- autoRebalanceScheduler.shutdown()
- deleteTopicManager.shutdown()
Utils.unregisterMBean(KafkaController.MBeanName)
+ deleteTopicManager.shutdown()
partitionStateMachine.shutdown()
replicaStateMachine.shutdown()
+ if(config.autoLeaderRebalanceEnable)
+ autoRebalanceScheduler.shutdown()
if(controllerContext.controllerChannelManager != null) {
controllerContext.controllerChannelManager.shutdown()
controllerContext.controllerChannelManager = null
+ info("Controller shutdown complete")
}
}
}
@@ -640,15 +641,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
def shutdown() = {
inLock(controllerContext.controllerLock) {
isRunning = false
- partitionStateMachine.shutdown()
- replicaStateMachine.shutdown()
- if (config.autoLeaderRebalanceEnable)
- autoRebalanceScheduler.shutdown()
- if(controllerContext.controllerChannelManager != null) {
- controllerContext.controllerChannelManager.shutdown()
- controllerContext.controllerChannelManager = null
- info("Controller shutdown complete")
- }
+ onControllerResignation()
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/39a56078/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
index 8262e10..6f615cf 100644
--- a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
+++ b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
@@ -22,6 +22,7 @@ import kafka.utils.Utils._
import collection.Set
import kafka.common.{ErrorMapping, TopicAndPartition}
import kafka.api.{StopReplicaResponse, RequestOrResponse}
+import java.util.concurrent.locks.ReentrantLock
/**
* This manages the state machine for topic deletion.
@@ -71,9 +72,10 @@ class TopicDeletionManager(controller: KafkaController,
val partitionStateMachine = controller.partitionStateMachine
val replicaStateMachine = controller.replicaStateMachine
var topicsToBeDeleted: mutable.Set[String] = mutable.Set.empty[String] ++ initialTopicsToBeDeleted
+ val deleteLock = new ReentrantLock()
var topicsIneligibleForDeletion: mutable.Set[String] = mutable.Set.empty[String] ++
(initialTopicsIneligibleForDeletion & initialTopicsToBeDeleted)
- val deleteTopicsCond = controllerContext.controllerLock.newCondition()
+ val deleteTopicsCond = deleteLock.newCondition()
var deleteTopicStateChanged: Boolean = false
var deleteTopicsThread: DeleteTopicsThread = null
val isDeleteTopicEnabled = controller.config.deleteTopicEnable
@@ -195,11 +197,14 @@ class TopicDeletionManager(controller: KafkaController,
* controllerLock should be acquired before invoking this API
*/
private def awaitTopicDeletionNotification() {
- while(!deleteTopicStateChanged) {
- info("Waiting for signal to start or continue topic deletion")
- deleteTopicsCond.await()
+ inLock(deleteLock) {
+ while(!deleteTopicStateChanged) {
+ info("Waiting for signal to start or continue topic deletion")
+
+ deleteTopicsCond.await()
+ }
+ deleteTopicStateChanged = false
}
- deleteTopicStateChanged = false
}
/**
@@ -207,7 +212,9 @@ class TopicDeletionManager(controller: KafkaController,
*/
private def resumeTopicDeletionThread() {
deleteTopicStateChanged = true
- deleteTopicsCond.signal()
+ inLock(deleteLock) {
+ deleteTopicsCond.signal()
+ }
}
/**
@@ -352,8 +359,9 @@ class TopicDeletionManager(controller: KafkaController,
class DeleteTopicsThread() extends ShutdownableThread("delete-topics-thread") {
val zkClient = controllerContext.zkClient
override def doWork() {
+ awaitTopicDeletionNotification()
+
inLock(controllerContext.controllerLock) {
- awaitTopicDeletionNotification()
val topicsQueuedForDeletion = Set.empty[String] ++ topicsToBeDeleted
if(topicsQueuedForDeletion.size > 0)
info("Handling deletion for topics " + topicsQueuedForDeletion.mkString(","))
http://git-wip-us.apache.org/repos/asf/kafka/blob/39a56078/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala
index 20fe93e..c7e058f 100644
--- a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala
@@ -96,5 +96,25 @@ class ServerShutdownTest extends JUnit3Suite with ZooKeeperTestHarness {
producer.close()
server.shutdown()
Utils.rm(server.config.logDirs)
+ verifyNonDaemonThreadsStatus
+ }
+
+ @Test
+ def testCleanShutdownWithDeleteTopicEnabled() {
+ val newProps = TestUtils.createBrokerConfig(0, port)
+ newProps.setProperty("delete.topic.enable", "true")
+ val newConfig = new KafkaConfig(newProps)
+ var server = new KafkaServer(newConfig)
+ server.startup()
+ server.shutdown()
+ server.awaitShutdown()
+ Utils.rm(server.config.logDirs)
+ verifyNonDaemonThreadsStatus
+ }
+
+ def verifyNonDaemonThreadsStatus() {
+ assertEquals(0, Thread.getAllStackTraces.keySet().toArray
+ .map(_.asInstanceOf[Thread])
+ .count(t => !t.isDaemon && t.isAlive && t.getClass.getCanonicalName.toLowerCase.startsWith("kafka")))
}
}