Repository: kafka
Updated Branches:
refs/heads/0.8.2 841d44139 -> c63e06200
kafka-1699; autoRebalanceScheduler.shutdown() causes deadlock while controller shutting down;
patched by Sriharsha Chintalapani; reviewed by Jun Rao
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/c63e0620
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/c63e0620
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/c63e0620
Branch: refs/heads/0.8.2
Commit: c63e06200d341345a1756e35ff0f48af2ec34d4a
Parents: 841d441
Author: Sriharsha Chintalapani <schintalapani@hortonworks.com>
Authored: Thu Oct 9 21:10:35 2014 -0700
Committer: Jun Rao <junrao@gmail.com>
Committed: Thu Oct 9 21:10:35 2014 -0700
----------------------------------------------------------------------
core/src/main/scala/kafka/controller/KafkaController.scala | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/c63e0620/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 381c52f..51a5bad 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -345,12 +345,13 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient,
val brokerSt
if (deleteTopicManager != null)
deleteTopicManager.shutdown()
+ // shutdown leader rebalance scheduler
+ if (config.autoLeaderRebalanceEnable)
+ autoRebalanceScheduler.shutdown()
+
inLock(controllerContext.controllerLock) {
// de-register partition ISR listener for on-going partition reassignment task
deregisterReassignedPartitionsIsrChangeListeners()
- // shutdown leader rebalance scheduler
- if (config.autoLeaderRebalanceEnable)
- autoRebalanceScheduler.shutdown()
// shutdown partition state machine
partitionStateMachine.shutdown()
// shutdown replica state machine
|