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/29 02:21:53 UTC
git commit: KAFKA-1317 follow up fix
Repository: kafka
Updated Branches:
refs/heads/0.8.1 39a560789 -> 5a6a1d83b
KAFKA-1317 follow up fix
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/5a6a1d83
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/5a6a1d83
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/5a6a1d83
Branch: refs/heads/0.8.1
Commit: 5a6a1d83b81caaebec0f10a0039b422acaeb5c35
Parents: 39a5607
Author: Neha Narkhede <ne...@gmail.com>
Authored: Fri Mar 28 18:21:47 2014 -0700
Committer: Neha Narkhede <ne...@gmail.com>
Committed: Fri Mar 28 18:21:47 2014 -0700
----------------------------------------------------------------------
.../scala/kafka/controller/TopicDeletionManager.scala | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/5a6a1d83/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 6f615cf..40c4c57 100644
--- a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
+++ b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
@@ -23,6 +23,7 @@ import collection.Set
import kafka.common.{ErrorMapping, TopicAndPartition}
import kafka.api.{StopReplicaResponse, RequestOrResponse}
import java.util.concurrent.locks.ReentrantLock
+import java.util.concurrent.atomic.AtomicBoolean
/**
* This manages the state machine for topic deletion.
@@ -76,7 +77,7 @@ class TopicDeletionManager(controller: KafkaController,
var topicsIneligibleForDeletion: mutable.Set[String] = mutable.Set.empty[String] ++
(initialTopicsIneligibleForDeletion & initialTopicsToBeDeleted)
val deleteTopicsCond = deleteLock.newCondition()
- var deleteTopicStateChanged: Boolean = false
+ var deleteTopicStateChanged: AtomicBoolean = new AtomicBoolean(false)
var deleteTopicsThread: DeleteTopicsThread = null
val isDeleteTopicEnabled = controller.config.deleteTopicEnable
@@ -86,7 +87,7 @@ class TopicDeletionManager(controller: KafkaController,
def start() {
if(isDeleteTopicEnabled) {
deleteTopicsThread = new DeleteTopicsThread()
- deleteTopicStateChanged = true
+ deleteTopicStateChanged.set(true)
deleteTopicsThread.start()
}
}
@@ -198,12 +199,10 @@ class TopicDeletionManager(controller: KafkaController,
*/
private def awaitTopicDeletionNotification() {
inLock(deleteLock) {
- while(!deleteTopicStateChanged) {
+ while(!deleteTopicStateChanged.compareAndSet(true, false)) {
info("Waiting for signal to start or continue topic deletion")
-
deleteTopicsCond.await()
}
- deleteTopicStateChanged = false
}
}
@@ -211,7 +210,7 @@ class TopicDeletionManager(controller: KafkaController,
* Signals the delete-topic-thread to process topic deletion
*/
private def resumeTopicDeletionThread() {
- deleteTopicStateChanged = true
+ deleteTopicStateChanged.set(true)
inLock(deleteLock) {
deleteTopicsCond.signal()
}